Skip to main content

nautilus_live/
runner.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//! Async event loop runner for live and sandbox trading nodes.
17//!
18//! `AsyncRunner` owns seven tokio mpsc channel pairs plus a shutdown
19//! signal channel. Construction creates the channels without side
20//! effects. The sender halves are placed into thread-local storage
21//! via [`AsyncRunner::bind_senders`] so that adapters and engine
22//! components can resolve them through the `get_*_sender()` accessors
23//! in `nautilus_common::runner` and `nautilus_common::live::runner`.
24//!
25//! Channel pairs:
26//!
27//! - **Time events**: timer callbacks dispatched by the clock.
28//! - **System events**: system notifications handled by the live node.
29//! - **System commands**: control requests handled by the live node.
30//! - **Execution events**: fills, order updates, and account state from
31//!   execution clients to the execution engine.
32//! - **Trading commands**: deferred order actions routed to their direct endpoint.
33//! - **Data events**: market data from adapters to the data engine.
34//! - **Data commands**: subscribe/unsubscribe requests to data clients.
35//!
36//! Both `AsyncRunner::run` and `LiveNode::run` use a `biased;` select with
37//! system and execution branches polled ahead of data branches. Within each
38//! channel pair, events are polled before commands.
39//!
40//! The runner can drive the event loop in two ways:
41//!
42//! - **Standalone**: call [`AsyncRunner::run`], which binds senders and
43//!   enters a `tokio::select!` loop internally.
44//! - **Integrated**: call [`AsyncRunner::take_channels`] to extract the
45//!   receivers and run the `select!` loop directly inside `LiveNode::run`,
46//!   where it is interleaved with startup, reconciliation, and shutdown
47//!   phases.
48//!
49//! # Invariants
50//!
51//! - `bind_senders` must be called before any code that reads from TLS.
52//!   This includes adapter constructors, clock initialization, and
53//!   execution client start methods. Every path from construction to
54//!   the event loop must bind before the first TLS read.
55//! - The event loop and all TLS consumers must execute on the same
56//!   thread. Senders are cloneable and `Send`, but the `RefCell`-backed
57//!   TLS slots are not accessible from other threads.
58//! - Only one runner at a time should own the TLS slots on a given
59//!   thread. `bind_senders` overwrites any existing TLS contents on the
60//!   thread, so the last caller wins.
61
62use std::{fmt::Debug, sync::Arc};
63
64use nautilus_common::{
65    live::runner::{
66        replace_data_event_sender, replace_exec_event_sender, replace_system_command_sender,
67        replace_system_event_sender,
68    },
69    messages::{
70        DataEvent, ExecutionEvent, ExecutionReport, SystemCommand, SystemEvent, data::DataCommand,
71        execution::TradingCommand,
72    },
73    msgbus::{self, MessagingSwitchboard},
74    runner::{
75        DataCommandSender, TimeEventMessage, TimeEventSender, TradingCommandMessage,
76        TradingCommandSender, replace_data_cmd_sender, replace_exec_cmd_sender,
77        replace_time_event_sender,
78    },
79};
80use nautilus_model::events::OrderEventAny;
81
82/// Asynchronous implementation of `DataCommandSender` for live environments.
83#[derive(Debug)]
84pub struct AsyncDataCommandSender {
85    cmd_tx: tokio::sync::mpsc::UnboundedSender<DataCommand>,
86}
87
88impl AsyncDataCommandSender {
89    #[must_use]
90    pub const fn new(cmd_tx: tokio::sync::mpsc::UnboundedSender<DataCommand>) -> Self {
91        Self { cmd_tx }
92    }
93}
94
95impl DataCommandSender for AsyncDataCommandSender {
96    fn execute(&self, command: DataCommand) {
97        if let Err(e) = self.cmd_tx.send(command) {
98            log::error!("Failed to send data command: {e}");
99        }
100    }
101}
102
103/// Asynchronous implementation of `TimeEventSender` for live environments.
104#[derive(Debug, Clone)]
105pub struct AsyncTimeEventSender {
106    time_tx: tokio::sync::mpsc::UnboundedSender<TimeEventMessage>,
107}
108
109impl AsyncTimeEventSender {
110    #[must_use]
111    pub const fn new(time_tx: tokio::sync::mpsc::UnboundedSender<TimeEventMessage>) -> Self {
112        Self { time_tx }
113    }
114}
115
116impl TimeEventSender for AsyncTimeEventSender {
117    fn send(&self, message: TimeEventMessage) {
118        if let Err(e) = self.time_tx.send(message) {
119            log::error!("Failed to send time event message: {e}");
120        }
121    }
122}
123
124/// Asynchronous implementation of `TradingCommandSender` for live environments.
125#[derive(Debug)]
126pub struct AsyncTradingCommandSender {
127    cmd_tx: tokio::sync::mpsc::UnboundedSender<TradingCommandMessage>,
128}
129
130impl AsyncTradingCommandSender {
131    #[must_use]
132    pub const fn new(cmd_tx: tokio::sync::mpsc::UnboundedSender<TradingCommandMessage>) -> Self {
133        Self { cmd_tx }
134    }
135}
136
137impl TradingCommandSender for AsyncTradingCommandSender {
138    fn execute(&self, message: TradingCommandMessage) {
139        if let Err(e) = self.cmd_tx.send(message) {
140            log::error!("Failed to send trading command: {e}");
141        }
142    }
143}
144
145pub trait Runner {
146    fn run(&mut self);
147}
148
149/// Channel receivers for the async event loop.
150///
151/// These can be extracted from `AsyncRunner` via `take_channels()` to drive
152/// the event loop directly on the same thread as the msgbus endpoints.
153#[derive(Debug)]
154pub struct AsyncRunnerChannels {
155    pub time_evt_rx: tokio::sync::mpsc::UnboundedReceiver<TimeEventMessage>,
156    pub system_evt_rx: tokio::sync::mpsc::UnboundedReceiver<SystemEvent>,
157    pub system_cmd_rx: tokio::sync::mpsc::UnboundedReceiver<SystemCommand>,
158    pub exec_evt_rx: tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
159    pub exec_cmd_rx: tokio::sync::mpsc::UnboundedReceiver<TradingCommandMessage>,
160    pub data_evt_rx: tokio::sync::mpsc::UnboundedReceiver<DataEvent>,
161    pub data_cmd_rx: tokio::sync::mpsc::UnboundedReceiver<DataCommand>,
162}
163
164#[cfg(feature = "node")]
165#[allow(
166    clippy::large_enum_variant,
167    reason = "runner events are consumed immediately; boxing would add routing allocations"
168)]
169pub(crate) enum PendingRunnerEvent {
170    TimeEvent(TimeEventMessage),
171    SystemEvent(SystemEvent),
172    SystemCommand(SystemCommand),
173    ExecEvent(ExecutionEvent),
174    ExecCommand(TradingCommandMessage),
175    DataEvent(DataEvent),
176    DataCommand(DataCommand),
177}
178
179pub struct AsyncRunner {
180    channels: AsyncRunnerChannels,
181    time_evt_tx: tokio::sync::mpsc::UnboundedSender<TimeEventMessage>,
182    system_evt_tx: tokio::sync::mpsc::UnboundedSender<SystemEvent>,
183    system_cmd_tx: tokio::sync::mpsc::UnboundedSender<SystemCommand>,
184    signal_rx: tokio::sync::mpsc::UnboundedReceiver<()>,
185    signal_tx: tokio::sync::mpsc::UnboundedSender<()>,
186    exec_evt_tx: tokio::sync::mpsc::UnboundedSender<ExecutionEvent>,
187    exec_cmd_tx: tokio::sync::mpsc::UnboundedSender<TradingCommandMessage>,
188    data_evt_tx: tokio::sync::mpsc::UnboundedSender<DataEvent>,
189    data_cmd_tx: tokio::sync::mpsc::UnboundedSender<DataCommand>,
190}
191
192/// Handle for stopping the `AsyncRunner` from another context.
193#[derive(Clone, Debug)]
194pub struct AsyncRunnerHandle {
195    signal_tx: tokio::sync::mpsc::UnboundedSender<()>,
196}
197
198impl AsyncRunnerHandle {
199    /// Signals the runner to stop.
200    pub fn stop(&self) {
201        if let Err(e) = self.signal_tx.send(()) {
202            log::error!("Failed to send shutdown signal: {e}");
203        }
204    }
205}
206
207impl Default for AsyncRunner {
208    fn default() -> Self {
209        Self::new()
210    }
211}
212
213impl Debug for AsyncRunner {
214    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
215        f.debug_struct(stringify!(AsyncRunner)).finish()
216    }
217}
218
219impl AsyncRunner {
220    /// Creates a new [`AsyncRunner`] instance.
221    ///
222    /// Creates channels but does not bind senders to thread-local storage.
223    /// Call [`bind_senders`](Self::bind_senders) before creating clients that
224    /// read from TLS, and again before entering the event loop.
225    #[must_use]
226    pub fn new() -> Self {
227        use tokio::sync::mpsc::unbounded_channel; // tokio-import-ok
228
229        let (time_evt_tx, time_evt_rx) = unbounded_channel::<TimeEventMessage>();
230        let (system_evt_tx, system_evt_rx) = unbounded_channel::<SystemEvent>();
231        let (system_cmd_tx, system_cmd_rx) = unbounded_channel::<SystemCommand>();
232        let (signal_tx, signal_rx) = unbounded_channel::<()>();
233        let (exec_evt_tx, exec_evt_rx) = unbounded_channel::<ExecutionEvent>();
234        let (exec_cmd_tx, exec_cmd_rx) = unbounded_channel::<TradingCommandMessage>();
235        let (data_evt_tx, data_evt_rx) = unbounded_channel::<DataEvent>();
236        let (data_cmd_tx, data_cmd_rx) = unbounded_channel::<DataCommand>();
237
238        Self {
239            channels: AsyncRunnerChannels {
240                time_evt_rx,
241                system_evt_rx,
242                system_cmd_rx,
243                exec_evt_rx,
244                exec_cmd_rx,
245                data_evt_rx,
246                data_cmd_rx,
247            },
248            time_evt_tx,
249            system_evt_tx,
250            system_cmd_tx,
251            signal_rx,
252            signal_tx,
253            exec_evt_tx,
254            exec_cmd_tx,
255            data_evt_tx,
256            data_cmd_tx,
257        }
258    }
259
260    /// Binds this runner's channel senders to thread-local storage.
261    ///
262    /// Call before creating clients that read from TLS (e.g., in the builder),
263    /// and again before entering the event loop to reclaim ownership if another
264    /// runner was constructed on this thread in the interim.
265    pub fn bind_senders(&self) {
266        replace_time_event_sender(Arc::new(AsyncTimeEventSender::new(
267            self.time_evt_tx.clone(),
268        )));
269        replace_system_event_sender(self.system_evt_tx.clone());
270        replace_system_command_sender(self.system_cmd_tx.clone());
271        replace_exec_event_sender(self.exec_evt_tx.clone());
272        replace_exec_cmd_sender(Arc::new(AsyncTradingCommandSender::new(
273            self.exec_cmd_tx.clone(),
274        )));
275        replace_data_event_sender(self.data_evt_tx.clone());
276        replace_data_cmd_sender(Arc::new(AsyncDataCommandSender::new(
277            self.data_cmd_tx.clone(),
278        )));
279    }
280
281    /// Stops the runner with an internal shutdown signal.
282    pub fn stop(&self) {
283        if let Err(e) = self.signal_tx.send(()) {
284            log::error!("Failed to send shutdown signal: {e}");
285        }
286    }
287
288    /// Returns a handle that can be used to stop the runner from another context.
289    #[must_use]
290    pub fn handle(&self) -> AsyncRunnerHandle {
291        AsyncRunnerHandle {
292            signal_tx: self.signal_tx.clone(),
293        }
294    }
295
296    /// Consumes the runner and returns the channel receivers for direct event loop driving.
297    ///
298    /// This is used when the event loop needs to run on the same thread as the msgbus
299    /// endpoints (which use thread-local storage).
300    #[must_use]
301    pub fn take_channels(self) -> AsyncRunnerChannels {
302        self.channels
303    }
304
305    /// Flushes all pending data events and commands from the channels.
306    ///
307    /// Loops until both data channels are empty, processing each item
308    /// into the cache immediately. Used in `start()` where channels are
309    /// not extracted.
310    pub fn flush_pending_data(&mut self) {
311        let mut total = 0;
312
313        loop {
314            let mut progressed = false;
315
316            // Events drain before commands here even though the runtime select
317            // prefers the opposite for everything-else: `LiveNode::start()`
318            // calls this after `connect_data_clients()` to push queued
319            // `DataEvent::Instrument` items into the cache. A pending
320            // subscription command (e.g. `SubscribeBars`) processed before the
321            // matching instrument lands would be rejected by the data engine.
322            while let Ok(evt) = self.channels.data_evt_rx.try_recv() {
323                Self::handle_data_event(evt);
324                progressed = true;
325                total += 1;
326            }
327
328            while let Ok(cmd) = self.channels.data_cmd_rx.try_recv() {
329                Self::handle_data_command(cmd);
330                progressed = true;
331                total += 1;
332            }
333
334            if !progressed {
335                break;
336            }
337        }
338
339        if total > 0 {
340            log::debug!("Flushed {total} pending data events/commands");
341        }
342    }
343
344    #[cfg(feature = "node")]
345    pub(crate) fn drain_pending_system_events(&mut self) -> Vec<SystemEvent> {
346        let mut events = Vec::new();
347
348        while let Ok(event) = self.channels.system_evt_rx.try_recv() {
349            events.push(event);
350        }
351
352        events
353    }
354
355    #[cfg(feature = "node")]
356    pub(crate) fn drain_pending_system_commands(&mut self) -> Vec<SystemCommand> {
357        let mut commands = Vec::new();
358
359        while let Ok(command) = self.channels.system_cmd_rx.try_recv() {
360            commands.push(command);
361        }
362
363        commands
364    }
365
366    /// Runs the async runner event loop.
367    ///
368    /// This method processes time, system, execution, and data events in an async loop.
369    /// It will run until a signal is received or the event streams are closed.
370    pub async fn run(&mut self) {
371        self.bind_senders();
372
373        log::info!("AsyncRunner starting");
374
375        loop {
376            tokio::select! {
377                biased;
378
379                Some(()) = self.signal_rx.recv() => {
380                    log::info!("AsyncRunner received signal, shutting down");
381                    return;
382                },
383                Some(handler) = self.channels.time_evt_rx.recv() => {
384                    let _ = Self::handle_time_event(handler);
385                },
386                Some(event) = self.channels.system_evt_rx.recv() => {
387                    log::error!("System event {event:?} requires the LiveNode runner");
388                },
389                Some(command) = self.channels.system_cmd_rx.recv() => {
390                    log::error!("System command {command:?} requires the LiveNode runner");
391                },
392                Some(evt) = self.channels.exec_evt_rx.recv() => {
393                    Self::handle_exec_event(evt);
394                },
395                Some(cmd) = self.channels.exec_cmd_rx.recv() => {
396                    Self::handle_trading_command(cmd);
397                },
398                Some(evt) = self.channels.data_evt_rx.recv() => {
399                    Self::handle_data_event(evt);
400                },
401                Some(cmd) = self.channels.data_cmd_rx.recv() => {
402                    Self::handle_data_command(cmd);
403                },
404                else => {
405                    log::debug!("AsyncRunner all channels closed, exiting");
406                    return;
407                }
408            };
409        }
410    }
411
412    /// Handles a time event by running its callback.
413    #[inline]
414    #[must_use]
415    pub fn handle_time_event(message: TimeEventMessage) -> bool {
416        message.dispatch()
417    }
418
419    /// Handles a data command by sending to the `DataEngine`.
420    #[inline]
421    pub fn handle_data_command(cmd: DataCommand) {
422        msgbus::send_data_command(MessagingSwitchboard::data_engine_execute(), cmd);
423    }
424
425    /// Handles a data event by sending to the appropriate `DataEngine` endpoint.
426    #[inline]
427    pub fn handle_data_event(event: DataEvent) {
428        match event {
429            DataEvent::Data(data) => {
430                msgbus::send_data(MessagingSwitchboard::data_engine_process_data(), data);
431            }
432            DataEvent::Instrument(data) => {
433                msgbus::send_any(MessagingSwitchboard::data_engine_process(), &data);
434            }
435            DataEvent::Response(resp) => {
436                msgbus::send_data_response(MessagingSwitchboard::data_engine_response(), resp);
437            }
438            DataEvent::FundingRate(funding_rate) => {
439                msgbus::send_any(MessagingSwitchboard::data_engine_process(), &funding_rate);
440            }
441            DataEvent::InstrumentStatus(status) => {
442                msgbus::send_any(MessagingSwitchboard::data_engine_process(), &status);
443            }
444            DataEvent::OptionGreeks(greeks) => {
445                msgbus::send_any(MessagingSwitchboard::data_engine_process(), &greeks);
446            }
447            #[cfg(feature = "defi")]
448            DataEvent::DeFi(data) => {
449                msgbus::send_defi_data(MessagingSwitchboard::data_engine_process_defi_data(), data);
450            }
451        }
452    }
453
454    /// Dispatches an internal execution command directly to the execution engine.
455    #[inline]
456    pub fn handle_exec_command(cmd: TradingCommand) {
457        msgbus::send_trading_command(MessagingSwitchboard::exec_engine_execute(), cmd);
458    }
459
460    /// Dispatches a deferred trading command to its direct endpoint.
461    #[inline]
462    pub fn handle_trading_command(message: TradingCommandMessage) {
463        let mut messages = vec![message];
464        while let Some(message) = messages.pop() {
465            messages.extend(message.dispatch().into_iter().rev());
466        }
467    }
468
469    /// Handles an execution event by sending to the appropriate engine endpoint.
470    #[inline]
471    pub fn handle_exec_event(event: ExecutionEvent) {
472        match event {
473            ExecutionEvent::Order(order_event) => {
474                msgbus::send_order_event(MessagingSwitchboard::exec_engine_process(), order_event);
475            }
476            ExecutionEvent::OrderSubmittedBatch(batch) => {
477                for submitted in batch {
478                    msgbus::send_order_event(
479                        MessagingSwitchboard::exec_engine_process(),
480                        OrderEventAny::Submitted(submitted),
481                    );
482                }
483            }
484            ExecutionEvent::OrderAcceptedBatch(batch) => {
485                for accepted in batch {
486                    msgbus::send_order_event(
487                        MessagingSwitchboard::exec_engine_process(),
488                        OrderEventAny::Accepted(accepted),
489                    );
490                }
491            }
492            ExecutionEvent::OrderCanceledBatch(batch) => {
493                for canceled in batch {
494                    msgbus::send_order_event(
495                        MessagingSwitchboard::exec_engine_process(),
496                        OrderEventAny::Canceled(canceled),
497                    );
498                }
499            }
500            ExecutionEvent::Report(report) => {
501                Self::handle_exec_report(report);
502            }
503            ExecutionEvent::Account(ref account) => {
504                msgbus::send_account_state(
505                    MessagingSwitchboard::portfolio_update_account(),
506                    account,
507                );
508            }
509        }
510    }
511
512    #[inline]
513    pub fn handle_exec_report(report: ExecutionReport) {
514        let endpoint = MessagingSwitchboard::exec_engine_reconcile_execution_report();
515        msgbus::send_execution_report(endpoint, report);
516    }
517}
518
519#[cfg(feature = "node")]
520impl AsyncRunner {
521    pub(crate) fn poll_pending(&mut self, mut process: impl FnMut(PendingRunnerEvent)) -> usize {
522        self.bind_senders();
523
524        let pending = (
525            self.channels.time_evt_rx.len(),
526            self.channels.system_evt_rx.len(),
527            self.channels.system_cmd_rx.len(),
528            self.channels.exec_evt_rx.len(),
529            self.channels.exec_cmd_rx.len(),
530            self.channels.data_evt_rx.len(),
531            self.channels.data_cmd_rx.len(),
532        );
533        let mut processed = 0;
534        processed += poll_channel(
535            &mut self.channels.time_evt_rx,
536            pending.0,
537            PendingRunnerEvent::TimeEvent,
538            &mut process,
539        );
540        processed += poll_channel(
541            &mut self.channels.system_evt_rx,
542            pending.1,
543            PendingRunnerEvent::SystemEvent,
544            &mut process,
545        );
546        processed += poll_channel(
547            &mut self.channels.system_cmd_rx,
548            pending.2,
549            PendingRunnerEvent::SystemCommand,
550            &mut process,
551        );
552        processed += poll_channel(
553            &mut self.channels.exec_evt_rx,
554            pending.3,
555            PendingRunnerEvent::ExecEvent,
556            &mut process,
557        );
558        processed += poll_channel(
559            &mut self.channels.exec_cmd_rx,
560            pending.4,
561            PendingRunnerEvent::ExecCommand,
562            &mut process,
563        );
564        processed += poll_channel(
565            &mut self.channels.data_evt_rx,
566            pending.5,
567            PendingRunnerEvent::DataEvent,
568            &mut process,
569        );
570        processed += poll_channel(
571            &mut self.channels.data_cmd_rx,
572            pending.6,
573            PendingRunnerEvent::DataCommand,
574            &mut process,
575        );
576        processed
577    }
578
579    pub(crate) async fn recv(&mut self) -> Option<PendingRunnerEvent> {
580        tokio::select! {
581            biased;
582
583            Some(message) = self.channels.time_evt_rx.recv() => {
584                Some(PendingRunnerEvent::TimeEvent(message))
585            }
586            Some(event) = self.channels.system_evt_rx.recv() => {
587                Some(PendingRunnerEvent::SystemEvent(event))
588            }
589            Some(command) = self.channels.system_cmd_rx.recv() => {
590                Some(PendingRunnerEvent::SystemCommand(command))
591            }
592            Some(event) = self.channels.exec_evt_rx.recv() => {
593                Some(PendingRunnerEvent::ExecEvent(event))
594            }
595            Some(command) = self.channels.exec_cmd_rx.recv() => {
596                Some(PendingRunnerEvent::ExecCommand(command))
597            }
598            Some(event) = self.channels.data_evt_rx.recv() => {
599                Some(PendingRunnerEvent::DataEvent(event))
600            }
601            Some(command) = self.channels.data_cmd_rx.recv() => {
602                Some(PendingRunnerEvent::DataCommand(command))
603            }
604            else => None,
605        }
606    }
607}
608
609#[cfg(feature = "node")]
610fn poll_channel<T>(
611    receiver: &mut tokio::sync::mpsc::UnboundedReceiver<T>,
612    pending: usize,
613    event: impl Fn(T) -> PendingRunnerEvent,
614    process: &mut impl FnMut(PendingRunnerEvent),
615) -> usize {
616    let mut processed = 0;
617
618    for _ in 0..pending {
619        let Ok(message) = receiver.try_recv() else {
620            break;
621        };
622
623        process(event(message));
624        processed += 1;
625    }
626
627    processed
628}
629
630#[cfg(test)]
631mod tests {
632    use std::{cell::RefCell, rc::Rc, sync::Arc, time::Duration};
633
634    use nautilus_common::{
635        cache::Cache,
636        clock::TestClock,
637        live::runner::{
638            get_data_event_sender, get_exec_event_sender, get_system_command_sender,
639            get_system_event_sender, try_get_system_command_sender, try_get_system_event_sender,
640        },
641        messages::{
642            ExecutionEvent, ExecutionReport,
643            data::{SubscribeCommand, SubscribeCustomData},
644            execution::{CancelAllOrders, TradingCommand},
645            system::{ReconnectSocket, SocketState, SocketStateChange},
646        },
647        msgbus::{TypedIntoHandler, stubs::get_typed_into_message_saving_handler},
648        runner::{
649            TimeEventMessage, get_data_cmd_sender, get_time_event_sender, get_trading_cmd_sender,
650            replace_exec_cmd_sender, try_get_time_event_sender, try_get_trading_cmd_sender,
651        },
652        timer::{TimeEvent, TimeEventCallback},
653    };
654    use nautilus_core::{UUID4, UnixNanos};
655    use nautilus_execution::engine::ExecutionEngine;
656    use nautilus_model::{
657        data::{Data, DataType, quote::QuoteTick},
658        enums::{
659            AccountType, LiquiditySide, OrderSide, OrderStatus, OrderType, PositionSide,
660            TimeInForce,
661        },
662        events::{
663            OrderAcceptedBatch, OrderCanceledBatch, OrderEvent, OrderEventAny, OrderSubmittedBatch,
664            account::state::AccountState,
665            order::spec::{OrderAcceptedSpec, OrderCanceledSpec, OrderSubmittedSpec},
666        },
667        identifiers::{
668            AccountId, ClientId, ClientOrderId, InstrumentId, PositionId, StrategyId, TradeId,
669            TraderId, Venue, VenueOrderId,
670        },
671        reports::{FillReport, OrderStatusReport, PositionStatusReport},
672        types::{Money, Price, Quantity},
673    };
674    use rstest::rstest;
675    use ustr::Ustr;
676
677    use super::*;
678
679    // Test fixture for creating test quotes
680    fn test_quote() -> QuoteTick {
681        QuoteTick {
682            instrument_id: InstrumentId::from("EUR/USD.SIM"),
683            bid_price: Price::from("1.10000"),
684            ask_price: Price::from("1.10001"),
685            bid_size: Quantity::from(1_000_000),
686            ask_size: Quantity::from(1_000_000),
687            ts_event: UnixNanos::default(),
688            ts_init: UnixNanos::default(),
689        }
690    }
691
692    fn test_system_event() -> SystemEvent {
693        SystemEvent::SocketState(SocketStateChange::new(
694            ClientId::from("BINANCE"),
695            Some(Venue::from("BINANCE")),
696            Ustr::from("binance-futures-market-streams"),
697            SocketState::Connected,
698        ))
699    }
700
701    fn test_system_command() -> SystemCommand {
702        SystemCommand::ReconnectSocket(ReconnectSocket::new(
703            TraderId::from("TRADER-001"),
704            ClientId::from("POLYMARKET"),
705            Ustr::from("polymarket-market-streams"),
706            UnixNanos::from(3),
707        ))
708    }
709
710    // Test fixture to create AsyncRunner with manual channels.
711    // Sender halves are dummies (not connected to the test receivers) since
712    // these tests exercise the event loop, not TLS binding.
713    fn create_test_runner(
714        time_evt_rx: tokio::sync::mpsc::UnboundedReceiver<TimeEventMessage>,
715        data_evt_rx: tokio::sync::mpsc::UnboundedReceiver<DataEvent>,
716        data_cmd_rx: tokio::sync::mpsc::UnboundedReceiver<DataCommand>,
717        exec_evt_rx: tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
718        exec_cmd_rx: tokio::sync::mpsc::UnboundedReceiver<TradingCommandMessage>,
719        signal_rx: tokio::sync::mpsc::UnboundedReceiver<()>,
720        signal_tx: tokio::sync::mpsc::UnboundedSender<()>,
721    ) -> AsyncRunner {
722        let (time_evt_tx, _) = tokio::sync::mpsc::unbounded_channel();
723        let (system_evt_tx, system_evt_rx) = tokio::sync::mpsc::unbounded_channel();
724        let (system_cmd_tx, system_cmd_rx) = tokio::sync::mpsc::unbounded_channel();
725        let (data_evt_tx, _) = tokio::sync::mpsc::unbounded_channel();
726        let (data_cmd_tx, _) = tokio::sync::mpsc::unbounded_channel();
727        let (exec_evt_tx, _) = tokio::sync::mpsc::unbounded_channel();
728        let (exec_cmd_tx, _) = tokio::sync::mpsc::unbounded_channel();
729
730        AsyncRunner {
731            channels: AsyncRunnerChannels {
732                time_evt_rx,
733                system_evt_rx,
734                system_cmd_rx,
735                exec_evt_rx,
736                exec_cmd_rx,
737                data_evt_rx,
738                data_cmd_rx,
739            },
740            time_evt_tx,
741            system_evt_tx,
742            system_cmd_tx,
743            exec_evt_tx,
744            exec_cmd_tx,
745            data_evt_tx,
746            data_cmd_tx,
747            signal_rx,
748            signal_tx,
749        }
750    }
751
752    #[cfg(feature = "node")]
753    #[rstest]
754    fn test_poll_pending_processes_entry_snapshot_across_channels() {
755        let (time_evt_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel();
756        let (data_evt_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel();
757        let (data_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel();
758        let (exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel();
759        let (exec_cmd_tx, exec_cmd_rx) = tokio::sync::mpsc::unbounded_channel();
760        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel();
761
762        let time_event = TimeEvent::new(
763            Ustr::from("test"),
764            UUID4::new(),
765            UnixNanos::from(1),
766            UnixNanos::from(2),
767        );
768        time_evt_tx
769            .send(TimeEventMessage::new(
770                time_event,
771                TimeEventCallback::from(|_: TimeEvent| {}),
772            ))
773            .unwrap();
774        exec_evt_tx
775            .send(ExecutionEvent::Order(OrderEventAny::Submitted(
776                OrderSubmittedSpec::builder()
777                    .client_order_id(ClientOrderId::from("O-POLL-001"))
778                    .build(),
779            )))
780            .unwrap();
781        exec_cmd_tx
782            .send(TradingCommandMessage::new(
783                MessagingSwitchboard::exec_engine_execute(),
784                TradingCommand::CancelAllOrders(CancelAllOrders::new(
785                    TraderId::from("TRADER-001"),
786                    None,
787                    StrategyId::from("S-POLL-001"),
788                    InstrumentId::from("EUR/USD.SIM"),
789                    Some(OrderSide::Buy),
790                    UUID4::new(),
791                    UnixNanos::from(3),
792                    None,
793                    None,
794                )),
795            ))
796            .unwrap();
797        data_evt_tx
798            .send(DataEvent::Data(Data::Quote(test_quote())))
799            .unwrap();
800        data_cmd_tx
801            .send(DataCommand::Subscribe(SubscribeCommand::Data(
802                SubscribeCustomData {
803                    client_id: Some(ClientId::from("POLL")),
804                    venue: None,
805                    data_type: DataType::new("QuoteTick", None, None),
806                    command_id: UUID4::new(),
807                    ts_init: UnixNanos::from(4),
808                    correlation_id: None,
809                    params: None,
810                },
811            )))
812            .unwrap();
813
814        let mut runner = create_test_runner(
815            time_evt_rx,
816            data_evt_rx,
817            data_cmd_rx,
818            exec_evt_rx,
819            exec_cmd_rx,
820            signal_rx,
821            signal_tx,
822        );
823        runner.bind_senders();
824        get_system_command_sender()
825            .send(test_system_command())
826            .unwrap();
827        get_system_event_sender().send(test_system_event()).unwrap();
828        get_system_event_sender().send(test_system_event()).unwrap();
829        let mut processed_by_channel = [0; 7];
830        let mut processed_order = Vec::new();
831
832        let first = runner.poll_pending(|event| match event {
833            PendingRunnerEvent::TimeEvent(_) => {
834                processed_by_channel[0] += 1;
835                processed_order.push("time");
836            }
837            PendingRunnerEvent::SystemEvent(_) => {
838                processed_by_channel[1] += 1;
839                processed_order.push("system_event");
840            }
841            PendingRunnerEvent::SystemCommand(_) => {
842                processed_by_channel[2] += 1;
843                processed_order.push("system_command");
844            }
845            PendingRunnerEvent::ExecEvent(_) => {
846                processed_by_channel[3] += 1;
847                processed_order.push("exec_event");
848            }
849            PendingRunnerEvent::ExecCommand(_) => {
850                processed_by_channel[4] += 1;
851                processed_order.push("exec_command");
852            }
853            PendingRunnerEvent::DataEvent(_) => {
854                processed_by_channel[5] += 1;
855                processed_order.push("data_event");
856                data_evt_tx
857                    .send(DataEvent::Data(Data::Quote(test_quote())))
858                    .unwrap();
859            }
860            PendingRunnerEvent::DataCommand(_) => {
861                processed_by_channel[6] += 1;
862                processed_order.push("data_command");
863            }
864        });
865        let second = runner.poll_pending(|event| match event {
866            PendingRunnerEvent::DataEvent(_) => {
867                processed_by_channel[5] += 1;
868                processed_order.push("data_event");
869            }
870            _ => panic!("Unexpected runner event"),
871        });
872
873        assert_eq!(first, 8);
874        assert_eq!(second, 1);
875        assert_eq!(processed_by_channel, [1, 2, 1, 1, 1, 2, 1]);
876        assert_eq!(
877            processed_order,
878            [
879                "time",
880                "system_event",
881                "system_event",
882                "system_command",
883                "exec_event",
884                "exec_command",
885                "data_event",
886                "data_command",
887                "data_event",
888            ]
889        );
890    }
891
892    #[cfg(feature = "node")]
893    #[tokio::test]
894    async fn test_recv_processes_system_event_before_command() {
895        let (_time_evt_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel();
896        let (_data_evt_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel();
897        let (_data_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel();
898        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel();
899        let (_exec_cmd_tx, exec_cmd_rx) = tokio::sync::mpsc::unbounded_channel();
900        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel();
901        let mut runner = create_test_runner(
902            time_evt_rx,
903            data_evt_rx,
904            data_cmd_rx,
905            exec_evt_rx,
906            exec_cmd_rx,
907            signal_rx,
908            signal_tx,
909        );
910
911        runner.system_cmd_tx.send(test_system_command()).unwrap();
912        runner.system_evt_tx.send(test_system_event()).unwrap();
913
914        assert!(matches!(
915            runner.recv().await,
916            Some(PendingRunnerEvent::SystemEvent(_))
917        ));
918        assert!(matches!(
919            runner.recv().await,
920            Some(PendingRunnerEvent::SystemCommand(_))
921        ));
922    }
923
924    #[rstest]
925    fn test_async_data_command_sender_creation() {
926        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
927        let sender = AsyncDataCommandSender::new(tx);
928        assert!(format!("{sender:?}").contains("AsyncDataCommandSender"));
929    }
930
931    #[rstest]
932    fn test_async_time_event_sender_creation() {
933        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
934        let sender = AsyncTimeEventSender::new(tx);
935        assert!(format!("{sender:?}").contains("AsyncTimeEventSender"));
936    }
937
938    #[tokio::test]
939    async fn test_async_data_command_sender_execute() {
940        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
941        let sender = AsyncDataCommandSender::new(tx);
942
943        let command = DataCommand::Subscribe(SubscribeCommand::Data(SubscribeCustomData {
944            client_id: Some(ClientId::from("TEST")),
945            venue: None,
946            data_type: DataType::new("QuoteTick", None, None),
947            command_id: UUID4::new(),
948            ts_init: UnixNanos::default(),
949            correlation_id: None,
950            params: None,
951        }));
952
953        sender.execute(command.clone());
954
955        let received = rx.recv().await.unwrap();
956        match (received, command) {
957            (
958                DataCommand::Subscribe(SubscribeCommand::Data(r)),
959                DataCommand::Subscribe(SubscribeCommand::Data(c)),
960            ) => {
961                assert_eq!(r.client_id, c.client_id);
962                assert_eq!(r.data_type, c.data_type);
963            }
964            _ => panic!("Command mismatch"),
965        }
966    }
967
968    #[tokio::test]
969    async fn test_async_time_event_sender_send() {
970        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
971        let sender = AsyncTimeEventSender::new(tx);
972
973        let event = TimeEvent::new(
974            Ustr::from("test"),
975            UUID4::new(),
976            UnixNanos::from(1),
977            UnixNanos::from(2),
978        );
979        let callback = TimeEventCallback::from(|_: TimeEvent| {});
980        let message = TimeEventMessage::new(event, callback);
981
982        sender.send(message);
983
984        assert!(rx.recv().await.is_some());
985    }
986
987    #[tokio::test]
988    async fn test_runner_shutdown_signal() {
989        // Create runner with manual channels to avoid global state
990        let (_data_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
991        let (_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
992        let (_time_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel::<TimeEventMessage>();
993        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
994        let (_exec_cmd_tx, exec_cmd_rx) =
995            tokio::sync::mpsc::unbounded_channel::<TradingCommandMessage>();
996        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
997
998        let mut runner = create_test_runner(
999            time_evt_rx,
1000            data_evt_rx,
1001            data_cmd_rx,
1002            exec_evt_rx,
1003            exec_cmd_rx,
1004            signal_rx,
1005            signal_tx.clone(),
1006        );
1007
1008        // Start runner
1009        let runner_handle = tokio::spawn(async move {
1010            runner.run().await;
1011        });
1012
1013        // Send shutdown signal
1014        signal_tx.send(()).unwrap();
1015
1016        // Runner should stop quickly
1017        let result = tokio::time::timeout(Duration::from_millis(100), runner_handle).await;
1018        assert!(result.is_ok(), "Runner should stop on signal");
1019    }
1020
1021    #[tokio::test]
1022    async fn test_runner_closes_on_channel_drop() {
1023        let (data_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
1024        let (_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
1025        let (_time_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel::<TimeEventMessage>();
1026        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1027        let (_exec_cmd_tx, exec_cmd_rx) =
1028            tokio::sync::mpsc::unbounded_channel::<TradingCommandMessage>();
1029        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1030
1031        let mut runner = create_test_runner(
1032            time_evt_rx,
1033            data_evt_rx,
1034            data_cmd_rx,
1035            exec_evt_rx,
1036            exec_cmd_rx,
1037            signal_rx,
1038            signal_tx.clone(),
1039        );
1040
1041        // Start runner
1042        let runner_handle = tokio::spawn(async move {
1043            runner.run().await;
1044        });
1045
1046        drop(data_tx);
1047
1048        // Yield to let runner enter event loop before stop signal
1049        tokio::task::yield_now().await;
1050        signal_tx.send(()).ok();
1051
1052        // Runner should stop when channels close or on signal
1053        let result = tokio::time::timeout(Duration::from_millis(200), runner_handle).await;
1054        assert!(
1055            result.is_ok(),
1056            "Runner should stop when channels close or on signal"
1057        );
1058    }
1059
1060    #[tokio::test]
1061    async fn test_concurrent_event_sending() {
1062        let (data_evt_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
1063        let (_data_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
1064        let (_time_evt_tx, time_evt_rx) =
1065            tokio::sync::mpsc::unbounded_channel::<TimeEventMessage>();
1066        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1067        let (_exec_cmd_tx, exec_cmd_rx) =
1068            tokio::sync::mpsc::unbounded_channel::<TradingCommandMessage>();
1069        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1070
1071        // Setup runner
1072        let mut runner = create_test_runner(
1073            time_evt_rx,
1074            data_evt_rx,
1075            data_cmd_rx,
1076            exec_evt_rx,
1077            exec_cmd_rx,
1078            signal_rx,
1079            signal_tx.clone(),
1080        );
1081
1082        // Spawn multiple concurrent senders
1083        let mut handles = vec![];
1084
1085        for _ in 0..5 {
1086            let tx_clone = data_evt_tx.clone();
1087
1088            let handle = tokio::spawn(async move {
1089                for _ in 0..20 {
1090                    let quote = test_quote();
1091                    tx_clone.send(DataEvent::Data(Data::Quote(quote))).unwrap();
1092                    tokio::task::yield_now().await;
1093                }
1094            });
1095            handles.push(handle);
1096        }
1097
1098        // Start runner in background
1099        let runner_handle = tokio::spawn(async move {
1100            runner.run().await;
1101        });
1102
1103        // Wait for all senders
1104        for handle in handles {
1105            handle.await.unwrap();
1106        }
1107
1108        // Yield to let runner enter event loop before stop signal
1109        tokio::task::yield_now().await;
1110        signal_tx.send(()).unwrap();
1111
1112        let _ = tokio::time::timeout(Duration::from_millis(200), runner_handle).await;
1113    }
1114
1115    #[rstest]
1116    #[case(10)]
1117    #[case(100)]
1118    #[case(1000)]
1119    fn test_channel_send_performance(#[case] count: usize) {
1120        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
1121        let quote = test_quote();
1122
1123        // Send events
1124        for _ in 0..count {
1125            tx.send(DataEvent::Data(Data::Quote(quote))).unwrap();
1126        }
1127
1128        // Verify all received
1129        let mut received = 0;
1130        while rx.try_recv().is_ok() {
1131            received += 1;
1132        }
1133
1134        assert_eq!(received, count);
1135    }
1136
1137    #[rstest]
1138    fn test_async_trading_command_sender_creation() {
1139        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
1140        let sender = AsyncTradingCommandSender::new(tx);
1141        assert!(format!("{sender:?}").contains("AsyncTradingCommandSender"));
1142    }
1143
1144    #[rstest]
1145    fn test_async_trading_command_sender_preserves_target_endpoints() {
1146        std::thread::spawn(|| {
1147            msgbus::get_message_bus().borrow_mut().dispose();
1148            let (risk_handler, risk_saving_handler) =
1149                get_typed_into_message_saving_handler::<TradingCommand>(Some(Ustr::from(
1150                    "RiskEngine.execute",
1151                )));
1152            msgbus::register_trading_command_endpoint(
1153                MessagingSwitchboard::risk_engine_execute(),
1154                risk_handler,
1155            );
1156            let (exec_handler, exec_saving_handler) =
1157                get_typed_into_message_saving_handler::<TradingCommand>(Some(Ustr::from(
1158                    "ExecEngine.execute",
1159                )));
1160            msgbus::register_trading_command_endpoint(
1161                MessagingSwitchboard::exec_engine_execute(),
1162                exec_handler,
1163            );
1164
1165            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<TradingCommandMessage>();
1166            let sender = AsyncTradingCommandSender::new(tx);
1167            sender.execute(TradingCommandMessage::new(
1168                MessagingSwitchboard::risk_engine_execute(),
1169                TradingCommand::CancelAllOrders(CancelAllOrders::new(
1170                    TraderId::from("TRADER-001"),
1171                    None,
1172                    StrategyId::from("RISK-001"),
1173                    InstrumentId::from("EUR/USD.SIM"),
1174                    Some(OrderSide::Buy),
1175                    UUID4::new(),
1176                    UnixNanos::default(),
1177                    None,
1178                    None,
1179                )),
1180            ));
1181            sender.execute(TradingCommandMessage::new(
1182                MessagingSwitchboard::exec_engine_execute(),
1183                TradingCommand::CancelAllOrders(CancelAllOrders::new(
1184                    TraderId::from("TRADER-001"),
1185                    None,
1186                    StrategyId::from("EXEC-001"),
1187                    InstrumentId::from("EUR/USD.SIM"),
1188                    Some(OrderSide::Sell),
1189                    UUID4::new(),
1190                    UnixNanos::default(),
1191                    None,
1192                    None,
1193                )),
1194            ));
1195
1196            AsyncRunner::handle_trading_command(rx.try_recv().unwrap());
1197            AsyncRunner::handle_trading_command(rx.try_recv().unwrap());
1198
1199            let risk_commands = risk_saving_handler.get_messages();
1200            let exec_commands = exec_saving_handler.get_messages();
1201            assert!(rx.try_recv().is_err());
1202            assert_eq!(risk_commands.len(), 1);
1203            assert_eq!(
1204                risk_commands[0].strategy_id(),
1205                Some(StrategyId::from("RISK-001"))
1206            );
1207            assert_eq!(exec_commands.len(), 1);
1208            assert_eq!(
1209                exec_commands[0].strategy_id(),
1210                Some(StrategyId::from("EXEC-001"))
1211            );
1212        })
1213        .join()
1214        .unwrap();
1215    }
1216
1217    #[rstest]
1218    fn test_async_runner_preserves_deferred_follow_up_order() {
1219        std::thread::spawn(|| {
1220            msgbus::get_message_bus().borrow_mut().dispose();
1221            let clock = Rc::new(RefCell::new(TestClock::new()));
1222            let cache = Rc::new(RefCell::new(Cache::default()));
1223            let exec_engine = Rc::new(RefCell::new(ExecutionEngine::new(clock, cache, None)));
1224            ExecutionEngine::register_msgbus_handlers(&exec_engine);
1225            msgbus::register_trading_command_endpoint(
1226                MessagingSwitchboard::risk_engine_execute(),
1227                TypedIntoHandler::from(|command: TradingCommand| {
1228                    msgbus::send_trading_command(
1229                        MessagingSwitchboard::exec_engine_queue_execute(),
1230                        command,
1231                    );
1232                }),
1233            );
1234            let (exec_handler, exec_saving_handler) =
1235                get_typed_into_message_saving_handler::<TradingCommand>(Some(Ustr::from(
1236                    "ExecEngine.execute",
1237                )));
1238            msgbus::register_trading_command_endpoint(
1239                MessagingSwitchboard::exec_engine_execute(),
1240                exec_handler,
1241            );
1242
1243            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<TradingCommandMessage>();
1244            let sender = Arc::new(AsyncTradingCommandSender::new(tx));
1245            replace_exec_cmd_sender(sender.clone());
1246            sender.execute(TradingCommandMessage::new(
1247                MessagingSwitchboard::risk_engine_execute(),
1248                TradingCommand::CancelAllOrders(CancelAllOrders::new(
1249                    TraderId::from("TRADER-001"),
1250                    None,
1251                    StrategyId::from("FIRST-001"),
1252                    InstrumentId::from("EUR/USD.SIM"),
1253                    Some(OrderSide::Buy),
1254                    UUID4::new(),
1255                    UnixNanos::default(),
1256                    None,
1257                    None,
1258                )),
1259            ));
1260            sender.execute(TradingCommandMessage::new(
1261                MessagingSwitchboard::exec_engine_execute(),
1262                TradingCommand::CancelAllOrders(CancelAllOrders::new(
1263                    TraderId::from("TRADER-001"),
1264                    None,
1265                    StrategyId::from("SECOND-001"),
1266                    InstrumentId::from("EUR/USD.SIM"),
1267                    Some(OrderSide::Sell),
1268                    UUID4::new(),
1269                    UnixNanos::default(),
1270                    None,
1271                    None,
1272                )),
1273            ));
1274
1275            AsyncRunner::handle_trading_command(rx.try_recv().unwrap());
1276            AsyncRunner::handle_trading_command(rx.try_recv().unwrap());
1277
1278            let commands = exec_saving_handler.get_messages();
1279            let strategy_ids = commands
1280                .iter()
1281                .map(TradingCommand::strategy_id)
1282                .collect::<Vec<_>>();
1283            assert!(rx.try_recv().is_err());
1284            assert_eq!(commands.len(), 2);
1285            assert_eq!(
1286                strategy_ids,
1287                vec![
1288                    Some(StrategyId::from("FIRST-001")),
1289                    Some(StrategyId::from("SECOND-001"))
1290                ]
1291            );
1292        })
1293        .join()
1294        .unwrap();
1295    }
1296
1297    #[rstest]
1298    fn test_async_runner_dispatches_deferred_exec_command_once() {
1299        std::thread::spawn(|| {
1300            msgbus::get_message_bus().borrow_mut().dispose();
1301            let clock = Rc::new(RefCell::new(TestClock::new()));
1302            let cache = Rc::new(RefCell::new(Cache::default()));
1303            let exec_engine = Rc::new(RefCell::new(ExecutionEngine::new(clock, cache, None)));
1304            ExecutionEngine::register_msgbus_handlers(&exec_engine);
1305
1306            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<TradingCommandMessage>();
1307            replace_exec_cmd_sender(Arc::new(AsyncTradingCommandSender::new(tx)));
1308            let command = TradingCommand::CancelAllOrders(CancelAllOrders::new(
1309                TraderId::from("TRADER-001"),
1310                None,
1311                StrategyId::from("EXEC-001"),
1312                InstrumentId::from("EUR/USD.SIM"),
1313                Some(OrderSide::Buy),
1314                UUID4::new(),
1315                UnixNanos::default(),
1316                None,
1317                None,
1318            ));
1319
1320            msgbus::send_trading_command(
1321                MessagingSwitchboard::exec_engine_queue_execute(),
1322                command,
1323            );
1324            assert_eq!(exec_engine.borrow().command_count(), 0);
1325
1326            AsyncRunner::handle_trading_command(rx.try_recv().unwrap());
1327
1328            assert!(rx.try_recv().is_err());
1329            assert_eq!(exec_engine.borrow().command_count(), 1);
1330        })
1331        .join()
1332        .unwrap();
1333    }
1334
1335    #[tokio::test]
1336    async fn test_runner_processes_trading_commands() {
1337        let (_data_evt_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
1338        let (_data_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
1339        let (_time_evt_tx, time_evt_rx) =
1340            tokio::sync::mpsc::unbounded_channel::<TimeEventMessage>();
1341        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1342        let (exec_cmd_tx, exec_cmd_rx) =
1343            tokio::sync::mpsc::unbounded_channel::<TradingCommandMessage>();
1344        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1345
1346        let mut runner = create_test_runner(
1347            time_evt_rx,
1348            data_evt_rx,
1349            data_cmd_rx,
1350            exec_evt_rx,
1351            exec_cmd_rx,
1352            signal_rx,
1353            signal_tx.clone(),
1354        );
1355
1356        let runner_handle = tokio::spawn(async move {
1357            runner.run().await;
1358        });
1359
1360        let command = TradingCommand::CancelAllOrders(CancelAllOrders::new(
1361            TraderId::from("TRADER-001"),
1362            None,
1363            StrategyId::from("S-001"),
1364            InstrumentId::from("EUR/USD.SIM"),
1365            Some(OrderSide::Buy),
1366            UUID4::new(),
1367            UnixNanos::default(),
1368            None,
1369            None, // correlation_id
1370        ));
1371        exec_cmd_tx
1372            .send(TradingCommandMessage::new(
1373                MessagingSwitchboard::exec_engine_execute(),
1374                command,
1375            ))
1376            .unwrap();
1377
1378        tokio::task::yield_now().await;
1379        signal_tx.send(()).unwrap();
1380
1381        let result = tokio::time::timeout(Duration::from_millis(100), runner_handle).await;
1382        assert!(result.is_ok(), "Runner should process command and stop");
1383    }
1384
1385    #[tokio::test]
1386    async fn test_runner_processes_multiple_trading_commands() {
1387        let (_data_evt_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
1388        let (_data_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
1389        let (_time_evt_tx, time_evt_rx) =
1390            tokio::sync::mpsc::unbounded_channel::<TimeEventMessage>();
1391        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1392        let (exec_cmd_tx, exec_cmd_rx) =
1393            tokio::sync::mpsc::unbounded_channel::<TradingCommandMessage>();
1394        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1395
1396        let mut runner = create_test_runner(
1397            time_evt_rx,
1398            data_evt_rx,
1399            data_cmd_rx,
1400            exec_evt_rx,
1401            exec_cmd_rx,
1402            signal_rx,
1403            signal_tx.clone(),
1404        );
1405
1406        let runner_handle = tokio::spawn(async move {
1407            runner.run().await;
1408        });
1409
1410        for i in 0..10 {
1411            let strategy_id = format!("S-{i:03}");
1412            let command = TradingCommand::CancelAllOrders(CancelAllOrders::new(
1413                TraderId::from("TRADER-001"),
1414                None,
1415                StrategyId::from(strategy_id.as_str()),
1416                InstrumentId::from("EUR/USD.SIM"),
1417                Some(OrderSide::Buy),
1418                UUID4::new(),
1419                UnixNanos::default(),
1420                None,
1421                None, // correlation_id
1422            ));
1423            exec_cmd_tx
1424                .send(TradingCommandMessage::new(
1425                    MessagingSwitchboard::exec_engine_execute(),
1426                    command,
1427                ))
1428                .unwrap();
1429        }
1430
1431        tokio::task::yield_now().await;
1432        signal_tx.send(()).unwrap();
1433
1434        let result = tokio::time::timeout(Duration::from_millis(100), runner_handle).await;
1435        assert!(
1436            result.is_ok(),
1437            "Runner should process all commands and stop"
1438        );
1439    }
1440
1441    #[tokio::test]
1442    async fn test_execution_event_order_channel() {
1443        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1444
1445        let event = OrderSubmittedSpec::builder()
1446            .client_order_id(ClientOrderId::from("O-001"))
1447            .build();
1448
1449        tx.send(ExecutionEvent::Order(OrderEventAny::Submitted(event)))
1450            .unwrap();
1451
1452        let received = rx.recv().await.unwrap();
1453        match received {
1454            ExecutionEvent::Order(OrderEventAny::Submitted(e)) => {
1455                assert_eq!(e.client_order_id(), ClientOrderId::from("O-001"));
1456            }
1457            _ => panic!("Expected OrderSubmitted event"),
1458        }
1459    }
1460
1461    #[tokio::test]
1462    async fn test_execution_report_order_status_channel() {
1463        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1464
1465        let report = OrderStatusReport::new(
1466            AccountId::from("SIM-001"),
1467            InstrumentId::from("EUR/USD.SIM"),
1468            Some(ClientOrderId::from("O-001")),
1469            VenueOrderId::from("V-001"),
1470            OrderSide::Buy.into(),
1471            OrderType::Market,
1472            TimeInForce::Gtc,
1473            OrderStatus::Accepted,
1474            Quantity::from(100_000),
1475            Quantity::from(100_000),
1476            UnixNanos::from(1),
1477            UnixNanos::from(2),
1478            UnixNanos::from(3),
1479            None,
1480        );
1481
1482        tx.send(ExecutionEvent::Report(ExecutionReport::Order(Box::new(
1483            report,
1484        ))))
1485        .unwrap();
1486
1487        let received = rx.recv().await.unwrap();
1488        match received {
1489            ExecutionEvent::Report(ExecutionReport::Order(r)) => {
1490                assert_eq!(r.venue_order_id.as_str(), "V-001");
1491                assert_eq!(r.order_status, OrderStatus::Accepted);
1492            }
1493            _ => panic!("Expected OrderStatusReport"),
1494        }
1495    }
1496
1497    #[tokio::test]
1498    async fn test_execution_report_fill() {
1499        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1500
1501        let report = FillReport::new(
1502            AccountId::from("SIM-001"),
1503            InstrumentId::from("EUR/USD.SIM"),
1504            VenueOrderId::from("V-001"),
1505            TradeId::from("T-001"),
1506            OrderSide::Buy,
1507            Quantity::from(100_000),
1508            Price::from("1.10000"),
1509            Money::from("10 USD"),
1510            LiquiditySide::Taker,
1511            Some(ClientOrderId::from("O-001")),
1512            None,
1513            UnixNanos::from(1),
1514            UnixNanos::from(2),
1515            None,
1516        );
1517
1518        tx.send(ExecutionEvent::Report(ExecutionReport::Fill(Box::new(
1519            report,
1520        ))))
1521        .unwrap();
1522
1523        let received = rx.recv().await.unwrap();
1524        match received {
1525            ExecutionEvent::Report(ExecutionReport::Fill(r)) => {
1526                assert_eq!(r.venue_order_id.as_str(), "V-001");
1527                assert_eq!(r.trade_id.to_string(), "T-001");
1528            }
1529            _ => panic!("Expected FillReport"),
1530        }
1531    }
1532
1533    #[tokio::test]
1534    async fn test_execution_report_position() {
1535        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1536
1537        let report = PositionStatusReport::new(
1538            AccountId::from("SIM-001"),
1539            InstrumentId::from("EUR/USD.SIM"),
1540            PositionSide::Long,
1541            Quantity::from(100_000),
1542            UnixNanos::from(1),
1543            UnixNanos::from(2),
1544            None,
1545            Some(PositionId::from("P-001")),
1546            None,
1547        );
1548
1549        tx.send(ExecutionEvent::Report(ExecutionReport::Position(Box::new(
1550            report,
1551        ))))
1552        .unwrap();
1553
1554        let received = rx.recv().await.unwrap();
1555        match received {
1556            ExecutionEvent::Report(ExecutionReport::Position(r)) => {
1557                assert_eq!(r.venue_position_id.unwrap().as_str(), "P-001");
1558            }
1559            _ => panic!("Expected PositionStatusReport"),
1560        }
1561    }
1562
1563    #[tokio::test]
1564    async fn test_execution_event_account() {
1565        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1566
1567        let account_state = AccountState::new(
1568            AccountId::from("SIM-001"),
1569            AccountType::Cash,
1570            vec![],
1571            vec![],
1572            true,
1573            UUID4::new(),
1574            UnixNanos::from(1),
1575            UnixNanos::from(2),
1576            None,
1577        );
1578
1579        tx.send(ExecutionEvent::Account(account_state)).unwrap();
1580
1581        let received = rx.recv().await.unwrap();
1582        match received {
1583            ExecutionEvent::Account(r) => {
1584                assert_eq!(r.account_id.as_str(), "SIM-001");
1585            }
1586            _ => panic!("Expected AccountState"),
1587        }
1588    }
1589
1590    #[tokio::test]
1591    async fn test_runner_stop_method() {
1592        let (_data_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
1593        let (_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
1594        let (_time_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel::<TimeEventMessage>();
1595        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1596        let (_exec_cmd_tx, exec_cmd_rx) =
1597            tokio::sync::mpsc::unbounded_channel::<TradingCommandMessage>();
1598        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1599
1600        let mut runner = create_test_runner(
1601            time_evt_rx,
1602            data_evt_rx,
1603            data_cmd_rx,
1604            exec_evt_rx,
1605            exec_cmd_rx,
1606            signal_rx,
1607            signal_tx.clone(),
1608        );
1609
1610        let runner_handle = tokio::spawn(async move {
1611            runner.run().await;
1612        });
1613
1614        // Use stop via signal_tx directly
1615        signal_tx.send(()).unwrap();
1616
1617        let result = tokio::time::timeout(Duration::from_millis(100), runner_handle).await;
1618        assert!(result.is_ok(), "Runner should stop when stop() is called");
1619    }
1620
1621    #[tokio::test]
1622    async fn test_all_event_types_integration() {
1623        let (data_evt_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
1624        let (data_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
1625        let (time_evt_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel::<TimeEventMessage>();
1626        let (exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1627        let (_exec_cmd_tx, exec_cmd_rx) =
1628            tokio::sync::mpsc::unbounded_channel::<TradingCommandMessage>();
1629        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1630
1631        let mut runner = create_test_runner(
1632            time_evt_rx,
1633            data_evt_rx,
1634            data_cmd_rx,
1635            exec_evt_rx,
1636            exec_cmd_rx,
1637            signal_rx,
1638            signal_tx.clone(),
1639        );
1640
1641        let runner_handle = tokio::spawn(async move {
1642            runner.run().await;
1643        });
1644
1645        // Send data event
1646        let quote = test_quote();
1647        data_evt_tx
1648            .send(DataEvent::Data(Data::Quote(quote)))
1649            .unwrap();
1650
1651        // Send data command
1652        let command = DataCommand::Subscribe(SubscribeCommand::Data(SubscribeCustomData {
1653            client_id: Some(ClientId::from("TEST")),
1654            venue: None,
1655            data_type: DataType::new("QuoteTick", None, None),
1656            command_id: UUID4::new(),
1657            ts_init: UnixNanos::default(),
1658            correlation_id: None,
1659            params: None,
1660        }));
1661        data_cmd_tx.send(command).unwrap();
1662
1663        // Send time event
1664        let event = TimeEvent::new(
1665            Ustr::from("test"),
1666            UUID4::new(),
1667            UnixNanos::from(1),
1668            UnixNanos::from(2),
1669        );
1670        let callback = TimeEventCallback::from(|_: TimeEvent| {});
1671        let message = TimeEventMessage::new(event, callback);
1672        time_evt_tx.send(message).unwrap();
1673
1674        // Send execution order event
1675        let order_event = OrderSubmittedSpec::builder()
1676            .client_order_id(ClientOrderId::from("O-001"))
1677            .build();
1678        exec_evt_tx
1679            .send(ExecutionEvent::Order(OrderEventAny::Submitted(order_event)))
1680            .unwrap();
1681
1682        // Send execution report (OrderStatus)
1683        let order_status = OrderStatusReport::new(
1684            AccountId::from("SIM-001"),
1685            InstrumentId::from("EUR/USD.SIM"),
1686            Some(ClientOrderId::from("O-001")),
1687            VenueOrderId::from("V-001"),
1688            OrderSide::Buy.into(),
1689            OrderType::Market,
1690            TimeInForce::Gtc,
1691            OrderStatus::Accepted,
1692            Quantity::from(100_000),
1693            Quantity::from(100_000),
1694            UnixNanos::from(1),
1695            UnixNanos::from(2),
1696            UnixNanos::from(3),
1697            None,
1698        );
1699        exec_evt_tx
1700            .send(ExecutionEvent::Report(ExecutionReport::Order(Box::new(
1701                order_status,
1702            ))))
1703            .unwrap();
1704
1705        // Send execution report (Fill)
1706        let fill = FillReport::new(
1707            AccountId::from("SIM-001"),
1708            InstrumentId::from("EUR/USD.SIM"),
1709            VenueOrderId::from("V-001"),
1710            TradeId::from("T-001"),
1711            OrderSide::Buy,
1712            Quantity::from(100_000),
1713            Price::from("1.10000"),
1714            Money::from("10 USD"),
1715            LiquiditySide::Taker,
1716            Some(ClientOrderId::from("O-001")),
1717            None,
1718            UnixNanos::from(1),
1719            UnixNanos::from(2),
1720            None,
1721        );
1722        exec_evt_tx
1723            .send(ExecutionEvent::Report(ExecutionReport::Fill(Box::new(
1724                fill,
1725            ))))
1726            .unwrap();
1727
1728        // Send execution report (Position)
1729        let position = PositionStatusReport::new(
1730            AccountId::from("SIM-001"),
1731            InstrumentId::from("EUR/USD.SIM"),
1732            PositionSide::Long,
1733            Quantity::from(100_000),
1734            UnixNanos::from(1),
1735            UnixNanos::from(2),
1736            None,
1737            Some(PositionId::from("P-001")),
1738            None,
1739        );
1740        exec_evt_tx
1741            .send(ExecutionEvent::Report(ExecutionReport::Position(Box::new(
1742                position,
1743            ))))
1744            .unwrap();
1745
1746        // Send account event
1747        let account_state = AccountState::new(
1748            AccountId::from("SIM-001"),
1749            AccountType::Cash,
1750            vec![],
1751            vec![],
1752            true,
1753            UUID4::new(),
1754            UnixNanos::from(1),
1755            UnixNanos::from(2),
1756            None,
1757        );
1758        exec_evt_tx
1759            .send(ExecutionEvent::Account(account_state))
1760            .unwrap();
1761
1762        // Yield to let runner enter event loop before stop signal
1763        tokio::task::yield_now().await;
1764        signal_tx.send(()).unwrap();
1765
1766        let result = tokio::time::timeout(Duration::from_millis(200), runner_handle).await;
1767        assert!(
1768            result.is_ok(),
1769            "Runner should process all event types and stop cleanly"
1770        );
1771    }
1772
1773    #[tokio::test]
1774    async fn test_runner_handle_stops_runner() {
1775        let (_data_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
1776        let (_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
1777        let (_time_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel::<TimeEventMessage>();
1778        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1779        let (_exec_cmd_tx, exec_cmd_rx) =
1780            tokio::sync::mpsc::unbounded_channel::<TradingCommandMessage>();
1781        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1782
1783        let mut runner = create_test_runner(
1784            time_evt_rx,
1785            data_evt_rx,
1786            data_cmd_rx,
1787            exec_evt_rx,
1788            exec_cmd_rx,
1789            signal_rx,
1790            signal_tx.clone(),
1791        );
1792
1793        // Get handle before moving runner
1794        let handle = runner.handle();
1795
1796        let runner_handle = tokio::spawn(async move {
1797            runner.run().await;
1798        });
1799
1800        // Use handle to stop
1801        handle.stop();
1802
1803        let result = tokio::time::timeout(Duration::from_millis(100), runner_handle).await;
1804        assert!(result.is_ok(), "Runner should stop via handle");
1805    }
1806
1807    #[tokio::test]
1808    async fn test_runner_handle_is_cloneable() {
1809        let (signal_tx, _signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1810        let handle = AsyncRunnerHandle { signal_tx };
1811
1812        let handle2 = handle.clone();
1813
1814        // Both handles should be able to send stop signals
1815        assert!(handle.signal_tx.send(()).is_ok());
1816        assert!(handle2.signal_tx.send(()).is_ok());
1817    }
1818
1819    #[tokio::test]
1820    async fn test_runner_processes_events_before_stop() {
1821        let (data_evt_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
1822        let (_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<DataCommand>();
1823        let (_time_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel::<TimeEventMessage>();
1824        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
1825        let (_exec_cmd_tx, exec_cmd_rx) =
1826            tokio::sync::mpsc::unbounded_channel::<TradingCommandMessage>();
1827        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1828
1829        let mut runner = create_test_runner(
1830            time_evt_rx,
1831            data_evt_rx,
1832            data_cmd_rx,
1833            exec_evt_rx,
1834            exec_cmd_rx,
1835            signal_rx,
1836            signal_tx.clone(),
1837        );
1838
1839        let handle = runner.handle();
1840
1841        // Send events before starting runner
1842        for _ in 0..10 {
1843            let quote = test_quote();
1844            data_evt_tx
1845                .send(DataEvent::Data(Data::Quote(quote)))
1846                .unwrap();
1847        }
1848
1849        let runner_handle = tokio::spawn(async move {
1850            runner.run().await;
1851        });
1852
1853        // Yield to let runner enter event loop before stop signal
1854        tokio::task::yield_now().await;
1855        handle.stop();
1856
1857        let result = tokio::time::timeout(Duration::from_millis(200), runner_handle).await;
1858        assert!(result.is_ok(), "Runner should process events and stop");
1859    }
1860
1861    #[rstest]
1862    fn test_new_does_not_bind_tls() {
1863        std::thread::spawn(|| {
1864            let _runner = AsyncRunner::new();
1865            assert!(try_get_time_event_sender().is_none());
1866            assert!(try_get_system_command_sender().is_none());
1867            assert!(try_get_system_event_sender().is_none());
1868            assert!(try_get_trading_cmd_sender().is_none());
1869        })
1870        .join()
1871        .unwrap();
1872    }
1873
1874    #[rstest]
1875    fn test_bind_senders_routes_to_runner_channels() {
1876        std::thread::spawn(|| {
1877            let mut runner = AsyncRunner::new();
1878            runner.bind_senders();
1879
1880            get_data_cmd_sender().execute(DataCommand::Subscribe(SubscribeCommand::Data(
1881                SubscribeCustomData {
1882                    client_id: Some(ClientId::from("TEST")),
1883                    venue: None,
1884                    data_type: DataType::new("test", None, None),
1885                    command_id: UUID4::new(),
1886                    ts_init: UnixNanos::default(),
1887                    correlation_id: None,
1888                    params: None,
1889                },
1890            )));
1891            assert!(runner.channels.data_cmd_rx.try_recv().is_ok());
1892
1893            get_trading_cmd_sender().execute(TradingCommandMessage::new(
1894                MessagingSwitchboard::exec_engine_execute(),
1895                TradingCommand::CancelAllOrders(CancelAllOrders::new(
1896                    TraderId::from("TRADER-001"),
1897                    None,
1898                    StrategyId::from("S-001"),
1899                    InstrumentId::from("EUR/USD.SIM"),
1900                    Some(OrderSide::Buy),
1901                    UUID4::new(),
1902                    UnixNanos::default(),
1903                    None,
1904                    None, // correlation_id
1905                )),
1906            ));
1907            assert!(runner.channels.exec_cmd_rx.try_recv().is_ok());
1908
1909            let event = TimeEvent::new(
1910                Ustr::from("test"),
1911                UUID4::new(),
1912                UnixNanos::from(1),
1913                UnixNanos::from(2),
1914            );
1915            let callback = TimeEventCallback::from(|_: TimeEvent| {});
1916            get_time_event_sender().send(TimeEventMessage::new(event, callback));
1917            assert!(runner.channels.time_evt_rx.try_recv().is_ok());
1918
1919            get_system_event_sender().send(test_system_event()).unwrap();
1920            assert_eq!(
1921                runner.channels.system_evt_rx.try_recv().unwrap(),
1922                test_system_event()
1923            );
1924
1925            get_system_command_sender()
1926                .send(test_system_command())
1927                .unwrap();
1928            assert_eq!(
1929                runner.channels.system_cmd_rx.try_recv().unwrap(),
1930                test_system_command()
1931            );
1932
1933            get_data_event_sender()
1934                .send(DataEvent::Data(Data::Quote(test_quote())))
1935                .unwrap();
1936            assert!(runner.channels.data_evt_rx.try_recv().is_ok());
1937
1938            let account = AccountState::new(
1939                AccountId::from("SIM-001"),
1940                AccountType::Cash,
1941                vec![],
1942                vec![],
1943                true,
1944                UUID4::new(),
1945                UnixNanos::from(1),
1946                UnixNanos::from(2),
1947                None,
1948            );
1949            get_exec_event_sender()
1950                .send(ExecutionEvent::Account(account))
1951                .unwrap();
1952            assert!(runner.channels.exec_evt_rx.try_recv().is_ok());
1953        })
1954        .join()
1955        .unwrap();
1956    }
1957
1958    #[cfg(feature = "node")]
1959    #[rstest]
1960    fn test_drain_pending_system_events_keeps_data_events_separate() {
1961        std::thread::spawn(|| {
1962            let mut runner = AsyncRunner::new();
1963            runner.bind_senders();
1964            let system_event = test_system_event();
1965
1966            get_system_event_sender().send(system_event).unwrap();
1967            get_data_event_sender()
1968                .send(DataEvent::Data(Data::Quote(test_quote())))
1969                .unwrap();
1970
1971            let system_events = runner.drain_pending_system_events();
1972
1973            assert_eq!(system_events, vec![system_event]);
1974            assert!(runner.channels.system_evt_rx.try_recv().is_err());
1975            assert!(runner.channels.data_evt_rx.try_recv().is_ok());
1976        })
1977        .join()
1978        .unwrap();
1979    }
1980
1981    #[cfg(feature = "node")]
1982    #[rstest]
1983    fn test_drain_pending_system_commands_keeps_events_separate() {
1984        std::thread::spawn(|| {
1985            let mut runner = AsyncRunner::new();
1986            runner.bind_senders();
1987            let system_command = test_system_command();
1988
1989            get_system_command_sender().send(system_command).unwrap();
1990            get_system_event_sender().send(test_system_event()).unwrap();
1991
1992            let system_commands = runner.drain_pending_system_commands();
1993
1994            assert_eq!(system_commands, vec![system_command]);
1995            assert!(runner.channels.system_cmd_rx.try_recv().is_err());
1996            assert!(runner.channels.system_evt_rx.try_recv().is_ok());
1997        })
1998        .join()
1999        .unwrap();
2000    }
2001
2002    #[rstest]
2003    fn test_bind_senders_reclaims_tls_from_previous_runner() {
2004        std::thread::spawn(|| {
2005            let mut runner1 = AsyncRunner::new();
2006            runner1.bind_senders();
2007
2008            let mut runner2 = AsyncRunner::new();
2009            runner2.bind_senders();
2010
2011            get_data_cmd_sender().execute(DataCommand::Subscribe(SubscribeCommand::Data(
2012                SubscribeCustomData {
2013                    client_id: Some(ClientId::from("TEST")),
2014                    venue: None,
2015                    data_type: DataType::new("test", None, None),
2016                    command_id: UUID4::new(),
2017                    ts_init: UnixNanos::default(),
2018                    correlation_id: None,
2019                    params: None,
2020                },
2021            )));
2022
2023            assert!(runner2.channels.data_cmd_rx.try_recv().is_ok());
2024            assert!(runner1.channels.data_cmd_rx.try_recv().is_err());
2025        })
2026        .join()
2027        .unwrap();
2028    }
2029
2030    #[tokio::test]
2031    async fn test_execution_event_order_submitted_batch_channel() {
2032        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
2033
2034        let events = vec![
2035            OrderSubmittedSpec::builder()
2036                .client_order_id(ClientOrderId::from("O-001"))
2037                .build(),
2038            OrderSubmittedSpec::builder()
2039                .client_order_id(ClientOrderId::from("O-002"))
2040                .build(),
2041        ];
2042
2043        let batch = OrderSubmittedBatch::new(events);
2044        tx.send(ExecutionEvent::OrderSubmittedBatch(batch)).unwrap();
2045
2046        let received = rx.recv().await.unwrap();
2047        match received {
2048            ExecutionEvent::OrderSubmittedBatch(b) => {
2049                assert_eq!(b.len(), 2);
2050                assert_eq!(b.events[0].client_order_id, ClientOrderId::from("O-001"));
2051                assert_eq!(b.events[1].client_order_id, ClientOrderId::from("O-002"));
2052            }
2053            _ => panic!("Expected OrderSubmittedBatch event"),
2054        }
2055    }
2056
2057    #[tokio::test]
2058    async fn test_execution_event_order_accepted_batch_channel() {
2059        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
2060
2061        let events = vec![
2062            OrderAcceptedSpec::builder()
2063                .client_order_id(ClientOrderId::from("O-001"))
2064                .build(),
2065            OrderAcceptedSpec::builder()
2066                .client_order_id(ClientOrderId::from("O-002"))
2067                .build(),
2068        ];
2069
2070        let batch = OrderAcceptedBatch::new(events);
2071        tx.send(ExecutionEvent::OrderAcceptedBatch(batch)).unwrap();
2072
2073        let received = rx.recv().await.unwrap();
2074        match received {
2075            ExecutionEvent::OrderAcceptedBatch(b) => {
2076                assert_eq!(b.len(), 2);
2077                assert_eq!(b.events[0].client_order_id, ClientOrderId::from("O-001"));
2078                assert_eq!(b.events[1].client_order_id, ClientOrderId::from("O-002"));
2079            }
2080            _ => panic!("Expected OrderAcceptedBatch event"),
2081        }
2082    }
2083
2084    #[tokio::test]
2085    async fn test_execution_event_order_canceled_batch_channel() {
2086        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
2087
2088        let events = vec![
2089            OrderCanceledSpec::builder()
2090                .client_order_id(ClientOrderId::from("O-001"))
2091                .build(),
2092            OrderCanceledSpec::builder()
2093                .client_order_id(ClientOrderId::from("O-002"))
2094                .build(),
2095        ];
2096
2097        let batch = OrderCanceledBatch::new(events);
2098        tx.send(ExecutionEvent::OrderCanceledBatch(batch)).unwrap();
2099
2100        let received = rx.recv().await.unwrap();
2101        match received {
2102            ExecutionEvent::OrderCanceledBatch(b) => {
2103                assert_eq!(b.len(), 2);
2104                assert_eq!(b.events[0].client_order_id, ClientOrderId::from("O-001"));
2105                assert_eq!(b.events[1].client_order_id, ClientOrderId::from("O-002"));
2106            }
2107            _ => panic!("Expected OrderCanceledBatch event"),
2108        }
2109    }
2110}