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//! system or execution channel pair, events precede commands. Timer and data channels rotate.
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::{
63    fmt::Debug,
64    sync::Arc,
65    thread::{self, ThreadId},
66};
67
68use nautilus_common::{
69    live::{
70        dispatch::DispatchMessage,
71        runner::{
72            replace_data_event_sender, replace_exec_event_sender, replace_system_command_sender,
73            replace_system_event_sender,
74        },
75        sender::{DispatchSender, EventSender},
76    },
77    messages::{
78        DataEvent, ExecutionEvent, ExecutionReport, SystemCommand, SystemEvent, data::DataCommand,
79        execution::TradingCommand,
80    },
81    msgbus::{self, MessagingSwitchboard},
82    runner::{
83        DataCommandSender, TimeEventMessage, TimeEventSender, TradingCommandMessage,
84        TradingCommandSender, replace_data_cmd_sender, replace_exec_cmd_sender,
85        replace_time_event_sender,
86    },
87};
88use nautilus_model::events::OrderEventAny;
89
90use crate::dispatch::drain_callbacks;
91#[cfg(feature = "node")]
92use crate::node::{LiveNodeHandle, NodeState};
93
94/// Asynchronous implementation of `DataCommandSender` for live environments.
95#[derive(Debug)]
96pub struct AsyncDataCommandSender {
97    cmd_tx: tokio::sync::mpsc::UnboundedSender<DispatchMessage<DataCommand>>,
98    owner: ThreadId,
99    #[cfg(feature = "node")]
100    node_handle: Option<LiveNodeHandle>,
101}
102
103impl AsyncDataCommandSender {
104    /// Creates a sender owned by the calling thread.
105    ///
106    /// Construct it on the runtime thread so command sends capture callback roots.
107    #[must_use]
108    pub fn new(cmd_tx: tokio::sync::mpsc::UnboundedSender<DispatchMessage<DataCommand>>) -> Self {
109        Self {
110            cmd_tx,
111            owner: thread::current().id(),
112            #[cfg(feature = "node")]
113            node_handle: None,
114        }
115    }
116}
117
118impl DataCommandSender for AsyncDataCommandSender {
119    fn execute(&self, command: DataCommand) {
120        if let Err(e) = self.cmd_tx.send(DispatchMessage::new(command, self.owner)) {
121            // Disposal releases retained subscriptions after the node drops its receivers
122            #[cfg(feature = "node")]
123            if node_stopped(self.node_handle.as_ref()) {
124                return;
125            }
126
127            log::error!("Failed to send data command: {e}");
128        }
129    }
130}
131
132/// Asynchronous implementation of `TimeEventSender` for live environments.
133#[derive(Debug, Clone)]
134pub struct AsyncTimeEventSender {
135    time_tx: DispatchSender<TimeEventMessage>,
136    #[cfg(feature = "node")]
137    node_handle: Option<LiveNodeHandle>,
138}
139
140impl AsyncTimeEventSender {
141    /// Creates a sender owned by the calling runtime thread.
142    #[must_use]
143    pub fn new(
144        time_tx: tokio::sync::mpsc::UnboundedSender<DispatchMessage<TimeEventMessage>>,
145    ) -> Self {
146        Self {
147            time_tx: DispatchSender::new(time_tx),
148            #[cfg(feature = "node")]
149            node_handle: None,
150        }
151    }
152}
153
154impl TimeEventSender for AsyncTimeEventSender {
155    fn send(&self, message: TimeEventMessage) {
156        if let Err(e) = self.time_tx.send(message) {
157            #[cfg(feature = "node")]
158            if node_stopped(self.node_handle.as_ref()) {
159                return;
160            }
161
162            log::error!("Failed to send time event message: {e}");
163        }
164    }
165}
166
167/// Asynchronous implementation of `TradingCommandSender` for live environments.
168#[derive(Debug)]
169pub struct AsyncTradingCommandSender {
170    cmd_tx: tokio::sync::mpsc::UnboundedSender<DispatchMessage<TradingCommandMessage>>,
171    owner: ThreadId,
172}
173
174impl AsyncTradingCommandSender {
175    /// Creates a sender owned by the calling thread.
176    ///
177    /// Construct it on the runtime thread so command sends capture callback roots.
178    #[must_use]
179    pub fn new(
180        cmd_tx: tokio::sync::mpsc::UnboundedSender<DispatchMessage<TradingCommandMessage>>,
181    ) -> Self {
182        Self {
183            cmd_tx,
184            owner: thread::current().id(),
185        }
186    }
187}
188
189impl TradingCommandSender for AsyncTradingCommandSender {
190    fn execute(&self, message: TradingCommandMessage) {
191        if let Err(e) = self.cmd_tx.send(DispatchMessage::new(message, self.owner)) {
192            log::error!("Failed to send trading command: {e}");
193        }
194    }
195}
196
197pub trait Runner {
198    fn run(&mut self);
199}
200
201/// Channel receivers for the async event loop.
202///
203/// These can be extracted from `AsyncRunner` via `take_channels()` to drive
204/// the event loop directly on the same thread as the msgbus endpoints.
205#[derive(Debug)]
206pub struct AsyncRunnerChannels {
207    pub time_evt_rx: tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<TimeEventMessage>>,
208    pub system_evt_rx: tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<SystemEvent>>,
209    pub system_cmd_rx: tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<SystemCommand>>,
210    pub exec_evt_rx: tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<ExecutionEvent>>,
211    pub exec_cmd_rx: tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<TradingCommandMessage>>,
212    pub data_evt_rx: tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<DataEvent>>,
213    pub data_cmd_rx: tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<DataCommand>>,
214}
215
216#[cfg(feature = "node")]
217#[allow(
218    clippy::large_enum_variant,
219    reason = "runner events are consumed immediately; boxing would add routing allocations"
220)]
221pub(crate) enum PendingRunnerEvent {
222    TimeEvent(DispatchMessage<TimeEventMessage>),
223    SystemEvent(DispatchMessage<SystemEvent>),
224    SystemCommand(DispatchMessage<SystemCommand>),
225    ExecEvent(DispatchMessage<ExecutionEvent>),
226    ExecCommand(DispatchMessage<TradingCommandMessage>),
227    DataEvent(DispatchMessage<DataEvent>),
228    DataCommand(DispatchMessage<DataCommand>),
229}
230
231pub struct AsyncRunner {
232    channels: AsyncRunnerChannels,
233    skip_time: bool,
234    skip_data: bool,
235    time_evt_tx: tokio::sync::mpsc::UnboundedSender<DispatchMessage<TimeEventMessage>>,
236    system_evt_tx: tokio::sync::mpsc::UnboundedSender<DispatchMessage<SystemEvent>>,
237    system_cmd_tx: tokio::sync::mpsc::UnboundedSender<DispatchMessage<SystemCommand>>,
238    signal_rx: tokio::sync::mpsc::UnboundedReceiver<()>,
239    signal_tx: tokio::sync::mpsc::UnboundedSender<()>,
240    exec_evt_tx: tokio::sync::mpsc::UnboundedSender<DispatchMessage<ExecutionEvent>>,
241    exec_cmd_tx: tokio::sync::mpsc::UnboundedSender<DispatchMessage<TradingCommandMessage>>,
242    data_evt_tx: tokio::sync::mpsc::UnboundedSender<DispatchMessage<DataEvent>>,
243    data_cmd_tx: tokio::sync::mpsc::UnboundedSender<DispatchMessage<DataCommand>>,
244}
245
246/// Handle for stopping the `AsyncRunner` from another context.
247#[derive(Clone, Debug)]
248pub struct AsyncRunnerHandle {
249    signal_tx: tokio::sync::mpsc::UnboundedSender<()>,
250}
251
252impl AsyncRunnerHandle {
253    /// Signals the runner to stop.
254    pub fn stop(&self) {
255        if let Err(e) = self.signal_tx.send(()) {
256            log::error!("Failed to send shutdown signal: {e}");
257        }
258    }
259}
260
261impl Default for AsyncRunner {
262    fn default() -> Self {
263        Self::new()
264    }
265}
266
267impl Debug for AsyncRunner {
268    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
269        f.debug_struct(stringify!(AsyncRunner)).finish()
270    }
271}
272
273impl AsyncRunner {
274    /// Creates a new [`AsyncRunner`] instance.
275    ///
276    /// Creates channels but does not bind senders to thread-local storage.
277    /// Call [`bind_senders`](Self::bind_senders) before creating clients that
278    /// read from TLS, and again before entering the event loop.
279    #[must_use]
280    #[rustfmt::skip]
281    pub fn new() -> Self {
282        use tokio::sync::mpsc::unbounded_channel; // tokio-import-ok
283
284        let (time_evt_tx, time_evt_rx) = unbounded_channel::<DispatchMessage<TimeEventMessage>>();
285        let (system_evt_tx, system_evt_rx) = unbounded_channel::<DispatchMessage<SystemEvent>>();
286        let (system_cmd_tx, system_cmd_rx) = unbounded_channel::<DispatchMessage<SystemCommand>>();
287        let (signal_tx, signal_rx) = unbounded_channel::<()>();
288        let (exec_evt_tx, exec_evt_rx) = unbounded_channel::<DispatchMessage<ExecutionEvent>>();
289        let (exec_cmd_tx, exec_cmd_rx) = unbounded_channel::<DispatchMessage<TradingCommandMessage>>();
290        let (data_evt_tx, data_evt_rx) = unbounded_channel::<DispatchMessage<DataEvent>>();
291        let (data_cmd_tx, data_cmd_rx) = unbounded_channel::<DispatchMessage<DataCommand>>();
292
293        Self {
294            channels: AsyncRunnerChannels {
295                time_evt_rx,
296                system_evt_rx,
297                system_cmd_rx,
298                exec_evt_rx,
299                exec_cmd_rx,
300                data_evt_rx,
301                data_cmd_rx,
302            },
303            skip_time: false,
304            skip_data: false,
305            time_evt_tx,
306            system_evt_tx,
307            system_cmd_tx,
308            signal_rx,
309            signal_tx,
310            exec_evt_tx,
311            exec_cmd_tx,
312            data_evt_tx,
313            data_cmd_tx,
314        }
315    }
316
317    /// Binds this runner's channel senders to thread-local storage.
318    ///
319    /// Call before creating clients that read from TLS (e.g., in the builder),
320    /// and again before entering the event loop to reclaim ownership if another
321    /// runner was constructed on this thread in the interim.
322    pub fn bind_senders(&self) {
323        self.bind_senders_with_data_sender(AsyncDataCommandSender::new(self.data_cmd_tx.clone()));
324    }
325
326    #[cfg(feature = "node")]
327    pub(crate) fn bind_senders_for_node(&self, handle: LiveNodeHandle) {
328        self.bind_senders_with_data_sender(AsyncDataCommandSender {
329            cmd_tx: self.data_cmd_tx.clone(),
330            owner: thread::current().id(),
331            node_handle: Some(handle),
332        });
333    }
334
335    #[rustfmt::skip]
336    fn bind_senders_with_data_sender(&self, sender: AsyncDataCommandSender) {
337        replace_time_event_sender(Arc::new(AsyncTimeEventSender {
338            time_tx: DispatchSender::new(self.time_evt_tx.clone()),
339            #[cfg(feature = "node")]
340            node_handle: sender.node_handle.clone(),
341        }));
342        replace_system_event_sender(EventSender::new(self.system_evt_tx.clone()));
343        replace_system_command_sender(DispatchSender::new(self.system_cmd_tx.clone()));
344        replace_exec_event_sender(EventSender::new(self.exec_evt_tx.clone()));
345        replace_exec_cmd_sender(Arc::new(AsyncTradingCommandSender::new(self.exec_cmd_tx.clone())));
346        replace_data_event_sender(EventSender::new(self.data_evt_tx.clone()));
347        replace_data_cmd_sender(Arc::new(sender));
348    }
349
350    /// Stops the runner with an internal shutdown signal.
351    pub fn stop(&self) {
352        if let Err(e) = self.signal_tx.send(()) {
353            log::error!("Failed to send shutdown signal: {e}");
354        }
355    }
356
357    /// Returns a handle that can be used to stop the runner from another context.
358    #[must_use]
359    pub fn handle(&self) -> AsyncRunnerHandle {
360        AsyncRunnerHandle {
361            signal_tx: self.signal_tx.clone(),
362        }
363    }
364
365    /// Consumes the runner and returns the channel receivers for direct event loop driving.
366    ///
367    /// This is used when the event loop needs to run on the same thread as the msgbus
368    /// endpoints (which use thread-local storage).
369    #[must_use]
370    pub fn take_channels(self) -> AsyncRunnerChannels {
371        self.channels
372    }
373
374    /// Flushes all pending data events and commands from the channels.
375    ///
376    /// Loops until both data channels are empty, processing each item
377    /// into the cache immediately. Used in `start()` where channels are
378    /// not extracted.
379    pub fn flush_pending_data(&mut self) {
380        let mut total = 0;
381
382        loop {
383            let mut progressed = false;
384
385            // Events drain before commands here even though the runtime select
386            // prefers the opposite for everything-else: `LiveNode::start()`
387            // calls this after `connect_data_clients()` to push queued
388            // instrument definitions into the cache. A pending
389            // subscription command (e.g. `SubscribeBars`) processed before the
390            // matching instrument lands would be rejected by the data engine.
391            while let Ok(evt) = self.channels.data_evt_rx.try_recv() {
392                Self::dispatch_data_event(evt);
393                progressed = true;
394                total += 1;
395            }
396
397            while let Ok(cmd) = self.channels.data_cmd_rx.try_recv() {
398                Self::handle_data_command(cmd);
399                progressed = true;
400                total += 1;
401            }
402
403            if !progressed {
404                break;
405            }
406        }
407
408        if total > 0 {
409            log::debug!("Flushed {total} pending data events/commands");
410        }
411    }
412
413    #[cfg(feature = "node")]
414    pub(crate) fn drain_pending_system_events(&mut self) -> Vec<DispatchMessage<SystemEvent>> {
415        let mut events = Vec::new();
416
417        while let Ok(event) = self.channels.system_evt_rx.try_recv() {
418            events.push(event);
419        }
420
421        events
422    }
423
424    #[cfg(feature = "node")]
425    pub(crate) fn drain_pending_system_commands(&mut self) -> Vec<DispatchMessage<SystemCommand>> {
426        let mut commands = Vec::new();
427
428        while let Ok(command) = self.channels.system_cmd_rx.try_recv() {
429            commands.push(command);
430        }
431
432        commands
433    }
434
435    /// Runs the async runner event loop.
436    ///
437    /// This method processes time, system, execution, and data events in an async loop.
438    /// It will run until a signal is received or the event streams are closed.
439    ///
440    /// # Errors
441    ///
442    /// Returns an error if callback dispatch fails. Pending channel messages and the failure latch
443    /// are retained.
444    pub async fn run(&mut self) -> anyhow::Result<()> {
445        self.bind_senders();
446
447        log::info!("AsyncRunner starting");
448
449        loop {
450            let callbacks_pending = drain_callbacks().await?;
451            let commands_empty = self.channels.data_cmd_rx.is_empty();
452            let poll_time =
453                !self.skip_time || (self.channels.data_evt_rx.is_empty() && commands_empty);
454            let poll_data = !self.skip_data || commands_empty;
455
456            tokio::select! {
457                biased;
458
459                Some(()) = self.signal_rx.recv() => {
460                    log::info!("AsyncRunner received signal, shutting down");
461                    return Ok(());
462                },
463                () = std::future::ready(()), if callbacks_pending => {},
464                Some(handler) = self.channels.time_evt_rx.recv(), if poll_time => {
465                    self.skip_time = true;
466                    let _ = Self::handle_time_event(handler);
467                },
468                Some(event) = self.channels.system_evt_rx.recv() => {
469                    self.skip_time = false;
470                    log::error!("System event {event} requires the LiveNode runner");
471                },
472                Some(command) = self.channels.system_cmd_rx.recv() => {
473                    self.skip_time = false;
474                    log::error!("System command {command} requires the LiveNode runner");
475                },
476                Some(evt) = self.channels.exec_evt_rx.recv() => {
477                    self.skip_time = false;
478                    Self::dispatch_exec_event(evt);
479                },
480                Some(cmd) = self.channels.exec_cmd_rx.recv() => {
481                    self.skip_time = false;
482                    Self::handle_trading_command(cmd);
483                },
484                Some(evt) = self.channels.data_evt_rx.recv(), if poll_data => {
485                    self.skip_time = false;
486                    self.skip_data = true;
487                    Self::dispatch_data_event(evt);
488                },
489                Some(cmd) = self.channels.data_cmd_rx.recv() => {
490                    self.skip_time = false;
491                    self.skip_data = false;
492                    Self::handle_data_command(cmd);
493                },
494                else => {
495                    log::debug!("AsyncRunner all channels closed, exiting");
496                    return Ok(());
497                }
498            };
499        }
500    }
501
502    /// Handles a time event under its originating callback root.
503    ///
504    /// # Panics
505    ///
506    /// Panics if a rooted message is processed outside its owner thread.
507    #[inline]
508    #[must_use]
509    pub fn handle_time_event(message: DispatchMessage<TimeEventMessage>) -> bool {
510        message.dispatch(TimeEventMessage::dispatch)
511    }
512
513    /// Handles a data command under its captured callback root by sending to the `DataEngine`.
514    ///
515    /// # Panics
516    ///
517    /// Panics if a rooted command is processed outside its owner thread.
518    #[inline]
519    pub fn handle_data_command(cmd: DispatchMessage<DataCommand>) {
520        cmd.dispatch(|cmd| {
521            msgbus::send_data_command(MessagingSwitchboard::data_engine_execute(), cmd);
522        });
523    }
524
525    /// Dispatches a received data event under its originating callback root.
526    ///
527    /// # Panics
528    ///
529    /// Panics if a rooted message is processed outside its owner thread.
530    pub fn dispatch_data_event(event: DispatchMessage<DataEvent>) {
531        event.dispatch(Self::handle_data_event);
532    }
533
534    /// Dispatches a received execution event under its originating callback root.
535    ///
536    /// # Panics
537    ///
538    /// Panics if a rooted message is processed outside its owner thread.
539    pub fn dispatch_exec_event(event: DispatchMessage<ExecutionEvent>) {
540        event.dispatch(Self::handle_exec_event);
541    }
542
543    /// Handles a data event by sending to the appropriate `DataEngine` endpoint.
544    #[inline]
545    pub fn handle_data_event(event: DataEvent) {
546        match event {
547            DataEvent::Data(data) => {
548                msgbus::send_data(MessagingSwitchboard::data_engine_process_data(), data);
549            }
550            DataEvent::ResponseEnvelope(response) => {
551                msgbus::send_any(MessagingSwitchboard::data_engine_process(), &response);
552            }
553            DataEvent::RequestFailed(failure) => {
554                msgbus::send_any(MessagingSwitchboard::data_engine_process(), &failure);
555            }
556            DataEvent::Response(resp) => {
557                msgbus::send_data_response(MessagingSwitchboard::data_engine_response(), resp);
558            }
559            #[cfg(feature = "defi")]
560            DataEvent::DeFi(data) => {
561                msgbus::send_defi_data(MessagingSwitchboard::data_engine_process_defi_data(), data);
562            }
563            #[cfg(feature = "defi")]
564            DataEvent::PoolSnapshotResponse(response) => {
565                msgbus::send_any(MessagingSwitchboard::data_engine_process(), &response);
566            }
567            #[cfg(not(feature = "defi"))]
568            #[allow(
569                unreachable_patterns,
570                reason = "DeFi variants can exist without this crate's defi feature"
571            )]
572            other => log::error!(
573                "Cannot handle {other} data event, nautilus-live built without its `defi` feature"
574            ),
575        }
576    }
577
578    /// Dispatches an internal execution command directly to the execution engine.
579    #[inline]
580    pub fn handle_exec_command(cmd: TradingCommand) {
581        msgbus::send_trading_command(MessagingSwitchboard::exec_engine_execute(), cmd);
582    }
583
584    /// Dispatches a trading command and its deferred children under their captured callback roots.
585    ///
586    /// # Panics
587    ///
588    /// Panics if a rooted command is processed outside its owner thread.
589    #[inline]
590    pub fn handle_trading_command(message: DispatchMessage<TradingCommandMessage>) {
591        message.dispatch_trading(|_| {});
592    }
593
594    /// Handles an execution event by sending to the appropriate engine endpoint.
595    #[inline]
596    pub fn handle_exec_event(event: ExecutionEvent) {
597        match event {
598            ExecutionEvent::Order(order_event) => {
599                msgbus::send_order_event(MessagingSwitchboard::exec_engine_process(), order_event);
600            }
601            ExecutionEvent::OrderSubmittedBatch(batch) => {
602                for submitted in batch {
603                    msgbus::send_order_event(
604                        MessagingSwitchboard::exec_engine_process(),
605                        OrderEventAny::Submitted(submitted),
606                    );
607                }
608            }
609            ExecutionEvent::OrderAcceptedBatch(batch) => {
610                for accepted in batch {
611                    msgbus::send_order_event(
612                        MessagingSwitchboard::exec_engine_process(),
613                        OrderEventAny::Accepted(accepted),
614                    );
615                }
616            }
617            ExecutionEvent::OrderCanceledBatch(batch) => {
618                for canceled in batch {
619                    msgbus::send_order_event(
620                        MessagingSwitchboard::exec_engine_process(),
621                        OrderEventAny::Canceled(canceled),
622                    );
623                }
624            }
625            ExecutionEvent::Report(report) => {
626                Self::handle_exec_report(report);
627            }
628            ExecutionEvent::Account(ref account) => {
629                msgbus::send_account_state(
630                    MessagingSwitchboard::portfolio_update_account(),
631                    account,
632                );
633            }
634        }
635    }
636
637    #[inline]
638    pub fn handle_exec_report(report: ExecutionReport) {
639        let endpoint = MessagingSwitchboard::exec_engine_reconcile_execution_report();
640        msgbus::send_execution_report(endpoint, report);
641    }
642}
643
644#[cfg(feature = "node")]
645impl AsyncRunner {
646    pub(crate) fn poll_pending(&mut self, mut process: impl FnMut(PendingRunnerEvent)) -> usize {
647        self.bind_senders();
648
649        let pending = (
650            self.channels.time_evt_rx.len(),
651            self.channels.system_evt_rx.len(),
652            self.channels.system_cmd_rx.len(),
653            self.channels.exec_evt_rx.len(),
654            self.channels.exec_cmd_rx.len(),
655            self.channels.data_evt_rx.len(),
656            self.channels.data_cmd_rx.len(),
657        );
658
659        let mut processed = 0;
660        processed += poll_channel(
661            &mut self.channels.time_evt_rx,
662            pending.0,
663            PendingRunnerEvent::TimeEvent,
664            &mut process,
665        );
666        processed += poll_channel(
667            &mut self.channels.system_evt_rx,
668            pending.1,
669            PendingRunnerEvent::SystemEvent,
670            &mut process,
671        );
672        processed += poll_channel(
673            &mut self.channels.system_cmd_rx,
674            pending.2,
675            PendingRunnerEvent::SystemCommand,
676            &mut process,
677        );
678        processed += poll_channel(
679            &mut self.channels.exec_evt_rx,
680            pending.3,
681            PendingRunnerEvent::ExecEvent,
682            &mut process,
683        );
684        processed += poll_channel(
685            &mut self.channels.exec_cmd_rx,
686            pending.4,
687            PendingRunnerEvent::ExecCommand,
688            &mut process,
689        );
690        processed += poll_channel(
691            &mut self.channels.data_evt_rx,
692            pending.5,
693            PendingRunnerEvent::DataEvent,
694            &mut process,
695        );
696        processed += poll_channel(
697            &mut self.channels.data_cmd_rx,
698            pending.6,
699            PendingRunnerEvent::DataCommand,
700            &mut process,
701        );
702        processed
703    }
704
705    pub(crate) async fn recv(&mut self) -> Option<PendingRunnerEvent> {
706        let commands_empty = self.channels.data_cmd_rx.is_empty();
707        let poll_time = !self.skip_time || (self.channels.data_evt_rx.is_empty() && commands_empty);
708        let poll_data = !self.skip_data || commands_empty;
709        tokio::select! {
710            biased;
711
712            Some(message) = self.channels.time_evt_rx.recv(), if poll_time => {
713                self.skip_time = true;
714                Some(PendingRunnerEvent::TimeEvent(message))
715            }
716            Some(event) = self.channels.system_evt_rx.recv() => {
717                self.skip_time = false;
718                Some(PendingRunnerEvent::SystemEvent(event))
719            }
720            Some(command) = self.channels.system_cmd_rx.recv() => {
721                self.skip_time = false;
722                Some(PendingRunnerEvent::SystemCommand(command))
723            }
724            Some(event) = self.channels.exec_evt_rx.recv() => {
725                self.skip_time = false;
726                Some(PendingRunnerEvent::ExecEvent(event))
727            }
728            Some(command) = self.channels.exec_cmd_rx.recv() => {
729                self.skip_time = false;
730                Some(PendingRunnerEvent::ExecCommand(command))
731            }
732            Some(event) = self.channels.data_evt_rx.recv(), if poll_data => {
733                self.skip_time = false;
734                self.skip_data = true;
735                Some(PendingRunnerEvent::DataEvent(event))
736            }
737            Some(command) = self.channels.data_cmd_rx.recv() => {
738                self.skip_time = false;
739                self.skip_data = false;
740                Some(PendingRunnerEvent::DataCommand(command))
741            }
742            else => None,
743        }
744    }
745}
746
747#[cfg(feature = "node")]
748fn node_stopped(handle: Option<&LiveNodeHandle>) -> bool {
749    handle.is_some_and(|handle| handle.state() == NodeState::Stopped)
750}
751
752#[cfg(feature = "node")]
753fn poll_channel<T>(
754    receiver: &mut tokio::sync::mpsc::UnboundedReceiver<T>,
755    pending: usize,
756    event: impl Fn(T) -> PendingRunnerEvent,
757    process: &mut impl FnMut(PendingRunnerEvent),
758) -> usize {
759    let mut processed = 0;
760
761    for _ in 0..pending {
762        let Ok(message) = receiver.try_recv() else {
763            break;
764        };
765
766        process(event(message));
767        processed += 1;
768    }
769
770    processed
771}
772
773#[cfg(test)]
774mod tests {
775    #[cfg(not(all(feature = "simulation", madsim)))]
776    use std::num::NonZeroU64;
777    use std::{cell::RefCell, rc::Rc, sync::Arc, time::Duration};
778
779    #[cfg(not(all(feature = "simulation", madsim)))]
780    use nautilus_common::live::LiveTimer;
781    use nautilus_common::{
782        actor,
783        cache::Cache,
784        clock::VirtualClock,
785        live::{
786            dispatch::DispatchMessage,
787            runner::{
788                get_data_event_sender, get_exec_event_sender, get_system_command_sender,
789                get_system_event_sender, try_get_system_command_sender,
790                try_get_system_event_sender,
791            },
792        },
793        messages::{
794            ExecutionEvent, ExecutionReport,
795            data::{SubscribeCommand, SubscribeCustomData},
796            execution::{CancelAllOrders, QueryAccount, TradingCommand},
797            system::{ReconnectSocket, SocketState, SocketStateChange},
798        },
799        msgbus::{TypedIntoHandler, stubs::get_typed_into_message_saving_handler},
800        runner::{
801            SyncTradingCommandSender, TimeEventMessage, drain_trading_cmd_queue,
802            get_data_cmd_sender, get_time_event_sender, get_trading_cmd_sender,
803            replace_exec_cmd_sender, try_get_time_event_sender, try_get_trading_cmd_sender,
804        },
805        timer::{TimeEvent, TimeEventCallback},
806    };
807    #[cfg(not(all(feature = "simulation", madsim)))]
808    use nautilus_core::time::get_atomic_clock_realtime;
809    use nautilus_core::{UUID4, UnixNanos};
810    use nautilus_execution::engine::ExecutionEngine;
811    use nautilus_model::{
812        data::{
813            Data, DataType, FundingRateUpdate, InstrumentStatus, OptionGreeks, quote::QuoteTick,
814            stubs::stub_instrument_status,
815        },
816        enums::{
817            AccountType, LiquiditySide, OrderSide, OrderStatus, OrderType, PositionSide,
818            TimeInForce,
819        },
820        events::{
821            OrderAcceptedBatch, OrderCanceledBatch, OrderEvent, OrderEventAny, OrderSubmittedBatch,
822            account::state::AccountState,
823            order::spec::{OrderAcceptedSpec, OrderCanceledSpec, OrderSubmittedSpec},
824        },
825        identifiers::{
826            AccountId, ClientId, ClientOrderId, InstrumentId, PositionId, StrategyId, TradeId,
827            TraderId, Venue, VenueOrderId,
828        },
829        instruments::{InstrumentAny, stubs::crypto_perpetual_ethusdt},
830        reports::{FillReport, OrderStatusReport, PositionStatusReport},
831        types::{Money, Price, Quantity},
832    };
833    use rstest::rstest;
834    use ustr::Ustr;
835
836    use super::*;
837
838    #[rstest]
839    fn test_runner_dispatches_typed_data_once_in_order() {
840        let mut runner = AsyncRunner::new();
841        let expected = vec![
842            Data::from(InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt())),
843            Data::FundingRate(FundingRateUpdate::new(
844                InstrumentId::from("ETHUSDT.BINANCE"),
845                "0.00012".parse().unwrap(),
846                Some(480),
847                Some(UnixNanos::from(300)),
848                UnixNanos::from(100),
849                UnixNanos::from(200),
850            )),
851            Data::InstrumentStatus(InstrumentStatus {
852                ts_event: UnixNanos::from(400),
853                ts_init: UnixNanos::from(500),
854                ..stub_instrument_status()
855            }),
856            Data::OptionGreeks(OptionGreeks {
857                instrument_id: InstrumentId::from("BTC-20270101-50000-C.DERIBIT"),
858                mark_iv: Some(0.55),
859                bid_iv: Some(0.54),
860                ask_iv: Some(0.56),
861                underlying_price: Some(50001.0),
862                open_interest: Some(123.0),
863                ts_event: UnixNanos::from(600),
864                ts_init: UnixNanos::from(700),
865                ..OptionGreeks::default()
866            }),
867        ];
868        let received = Rc::new(RefCell::new(Vec::<Data>::new()));
869        let observed = received.clone();
870        msgbus::register_data_endpoint(
871            MessagingSwitchboard::data_engine_process_data(),
872            TypedIntoHandler::from(move |data| observed.borrow_mut().push(data)),
873        );
874
875        for data in &expected {
876            runner
877                .data_evt_tx
878                .send(DataEvent::Data(data.clone()).into())
879                .unwrap();
880        }
881
882        runner.flush_pending_data();
883
884        assert_eq!(
885            serde_json::to_value(&*received.borrow()).unwrap(),
886            serde_json::to_value(&expected).unwrap(),
887        );
888    }
889
890    #[rstest]
891    #[case(false)]
892    #[case(true)]
893    #[tokio::test]
894    async fn test_runner_callback_failure_stops_before_next_command(#[case] before_run: bool) {
895        actor::clear_callbacks().unwrap();
896        let mut runner = AsyncRunner::new();
897        let received = Rc::new(RefCell::new(Vec::new()));
898        let observed = received.clone();
899        msgbus::register_trading_command_endpoint(
900            MessagingSwitchboard::risk_engine_execute(),
901            TypedIntoHandler::from(move |command: TradingCommand| {
902                observed.borrow_mut().push(command.ts_init());
903                crate::dispatch::tests::latch_callback_failure();
904            }),
905        );
906
907        let sender = AsyncTradingCommandSender::new(runner.exec_cmd_tx.clone());
908        let seed_endpoint = ustr::Ustr::from("test.callback-root-seed");
909        msgbus::register_trading_command_endpoint(
910            seed_endpoint.into(),
911            TypedIntoHandler::from(move |command: TradingCommand| {
912                sender.execute(TradingCommandMessage::new(
913                    MessagingSwitchboard::risk_engine_execute(),
914                    command,
915                ));
916            }),
917        );
918
919        for timestamp in [17, 23] {
920            SyncTradingCommandSender.execute(TradingCommandMessage::new(
921                seed_endpoint.into(),
922                TradingCommand::QueryAccount(QueryAccount::new(
923                    "TRADER-001".into(),
924                    None,
925                    "SIM-001".into(),
926                    UUID4::new(),
927                    timestamp.into(),
928                    None,
929                    None,
930                )),
931            ));
932        }
933
934        drain_trading_cmd_queue();
935        let first = runner.channels.exec_cmd_rx.try_recv().unwrap();
936        let second = runner.channels.exec_cmd_rx.try_recv().unwrap();
937        assert!(first.is_rooted());
938        assert!(second.is_rooted());
939        runner.exec_cmd_tx.send(first).unwrap();
940        runner.exec_cmd_tx.send(second).unwrap();
941
942        if before_run {
943            crate::dispatch::tests::latch_callback_failure();
944        }
945
946        let error = runner.run().await.unwrap_err();
947
948        assert_eq!(
949            error.downcast_ref::<actor::CallbackDispatchError>(),
950            Some(&actor::CallbackDispatchError::DeliveryUnwound)
951        );
952        assert_eq!(
953            *received.borrow(),
954            if before_run {
955                vec![]
956            } else {
957                vec![UnixNanos::from(17)]
958            }
959        );
960        assert_eq!(
961            runner.channels.exec_cmd_rx.len(),
962            if before_run { 2 } else { 1 }
963        );
964        assert!(!runner.exec_cmd_tx.is_closed());
965        assert_eq!(
966            actor::callback_failure(),
967            Some(actor::CallbackDispatchError::DeliveryUnwound)
968        );
969        drop(runner);
970        assert_eq!(actor::clear_callbacks(), Ok(()));
971    }
972
973    #[tokio::test]
974    async fn test_runner_stop_preserves_messages_and_channel_only_scheduling() {
975        let mut runner = AsyncRunner::new();
976        let received = Rc::new(RefCell::new(Vec::new()));
977        let stop = runner.signal_tx.clone();
978
979        for timestamp in 1..=65 {
980            let received = received.clone();
981            let stop = stop.clone();
982            runner
983                .time_evt_tx
984                .send(
985                    TimeEventMessage::new(
986                        TimeEvent::new(
987                            "runner-resume".into(),
988                            UUID4::new(),
989                            timestamp.into(),
990                            timestamp.into(),
991                        ),
992                        TimeEventCallback::RustLocal(Rc::new(move |_| {
993                            received.borrow_mut().push(timestamp);
994                            if timestamp == 65 {
995                                stop.send(()).unwrap();
996                            }
997                        })),
998                    )
999                    .into(),
1000                )
1001                .unwrap();
1002        }
1003
1004        stop.send(()).unwrap();
1005        runner.run().await.unwrap();
1006
1007        assert!(received.borrow().is_empty());
1008        assert_eq!(runner.channels.time_evt_rx.len(), 65);
1009        assert!(!runner.time_evt_tx.is_closed());
1010
1011        let (result, at_yield) = tokio::join!(biased;
1012            runner.run(),
1013            async { received.borrow().clone() },
1014        );
1015        result.unwrap();
1016
1017        let expected: Vec<u64> = (1..=65).collect();
1018        assert_eq!(at_yield, expected);
1019        assert_eq!(*received.borrow(), expected);
1020        assert_eq!(runner.channels.time_evt_rx.len(), 0);
1021        assert!(!runner.time_evt_tx.is_closed());
1022    }
1023
1024    // Test fixture for creating test quotes
1025    fn test_quote() -> QuoteTick {
1026        QuoteTick {
1027            instrument_id: InstrumentId::from("EUR/USD.SIM"),
1028            bid_price: Price::from("1.10000"),
1029            ask_price: Price::from("1.10001"),
1030            bid_size: Quantity::from(1_000_000),
1031            ask_size: Quantity::from(1_000_000),
1032            ts_event: UnixNanos::default(),
1033            ts_init: UnixNanos::default(),
1034        }
1035    }
1036
1037    fn test_system_event() -> SystemEvent {
1038        SystemEvent::SocketState(SocketStateChange::new(
1039            ClientId::from("BINANCE"),
1040            Some(Venue::from("BINANCE")),
1041            Ustr::from("binance-futures-market-streams"),
1042            SocketState::Connected,
1043        ))
1044    }
1045
1046    fn test_system_command() -> SystemCommand {
1047        SystemCommand::ReconnectSocket(ReconnectSocket::new(
1048            TraderId::from("TRADER-001"),
1049            ClientId::from("POLYMARKET"),
1050            Ustr::from("polymarket-market-streams"),
1051            UnixNanos::from(3),
1052        ))
1053    }
1054
1055    // Test fixture to create AsyncRunner with manual channels.
1056    // Sender halves are dummies (not connected to the test receivers) since
1057    // these tests exercise the event loop, not TLS binding.
1058    fn create_test_runner(
1059        time_evt_rx: tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<TimeEventMessage>>,
1060        data_evt_rx: tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<DataEvent>>,
1061        data_cmd_rx: tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<DataCommand>>,
1062        exec_evt_rx: tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<ExecutionEvent>>,
1063        exec_cmd_rx: tokio::sync::mpsc::UnboundedReceiver<DispatchMessage<TradingCommandMessage>>,
1064        signal_rx: tokio::sync::mpsc::UnboundedReceiver<()>,
1065        signal_tx: tokio::sync::mpsc::UnboundedSender<()>,
1066    ) -> AsyncRunner {
1067        let (time_evt_tx, _) = tokio::sync::mpsc::unbounded_channel();
1068        let (system_evt_tx, system_evt_rx) = tokio::sync::mpsc::unbounded_channel();
1069        let (system_cmd_tx, system_cmd_rx) = tokio::sync::mpsc::unbounded_channel();
1070        let (data_evt_tx, _) = tokio::sync::mpsc::unbounded_channel();
1071        let (data_cmd_tx, _) = tokio::sync::mpsc::unbounded_channel();
1072        let (exec_evt_tx, _) = tokio::sync::mpsc::unbounded_channel();
1073        let (exec_cmd_tx, _) = tokio::sync::mpsc::unbounded_channel();
1074
1075        AsyncRunner {
1076            channels: AsyncRunnerChannels {
1077                time_evt_rx,
1078                system_evt_rx,
1079                system_cmd_rx,
1080                exec_evt_rx,
1081                exec_cmd_rx,
1082                data_evt_rx,
1083                data_cmd_rx,
1084            },
1085            skip_time: false,
1086            skip_data: false,
1087            time_evt_tx,
1088            system_evt_tx,
1089            system_cmd_tx,
1090            exec_evt_tx,
1091            exec_cmd_tx,
1092            data_evt_tx,
1093            data_cmd_tx,
1094            signal_rx,
1095            signal_tx,
1096        }
1097    }
1098
1099    #[cfg(feature = "node")]
1100    #[rstest]
1101    fn test_poll_pending_processes_entry_snapshot_across_channels() {
1102        let (time_evt_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel();
1103        let (data_evt_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel();
1104        let (data_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel();
1105        let (exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel();
1106        let (exec_cmd_tx, exec_cmd_rx) = tokio::sync::mpsc::unbounded_channel();
1107        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel();
1108
1109        let time_event = TimeEvent::new(
1110            Ustr::from("test"),
1111            UUID4::new(),
1112            UnixNanos::from(1),
1113            UnixNanos::from(2),
1114        );
1115        time_evt_tx
1116            .send(
1117                (TimeEventMessage::new(time_event, TimeEventCallback::from(|_: TimeEvent| {})))
1118                    .into(),
1119            )
1120            .unwrap();
1121        exec_evt_tx
1122            .send(
1123                (ExecutionEvent::Order(OrderEventAny::Submitted(
1124                    OrderSubmittedSpec::builder()
1125                        .client_order_id(ClientOrderId::from("O-POLL-001"))
1126                        .build(),
1127                )))
1128                .into(),
1129            )
1130            .unwrap();
1131        exec_cmd_tx
1132            .send(
1133                TradingCommandMessage::new(
1134                    MessagingSwitchboard::exec_engine_execute(),
1135                    TradingCommand::CancelAllOrders(CancelAllOrders::new(
1136                        TraderId::from("TRADER-001"),
1137                        None,
1138                        StrategyId::from("S-POLL-001"),
1139                        InstrumentId::from("EUR/USD.SIM"),
1140                        Some(OrderSide::Buy),
1141                        UUID4::new(),
1142                        UnixNanos::from(3),
1143                        None,
1144                        None,
1145                    )),
1146                )
1147                .into(),
1148            )
1149            .unwrap();
1150        data_evt_tx
1151            .send((DataEvent::Data(Data::Quote(test_quote()))).into())
1152            .unwrap();
1153        data_cmd_tx
1154            .send(
1155                DataCommand::Subscribe(SubscribeCommand::Data(SubscribeCustomData {
1156                    client_id: Some(ClientId::from("POLL")),
1157                    venue: None,
1158                    data_type: DataType::new("QuoteTick", None, None),
1159                    command_id: UUID4::new(),
1160                    ts_init: UnixNanos::from(4),
1161                    correlation_id: None,
1162                    params: None,
1163                }))
1164                .into(),
1165            )
1166            .unwrap();
1167
1168        let mut runner = create_test_runner(
1169            time_evt_rx,
1170            data_evt_rx,
1171            data_cmd_rx,
1172            exec_evt_rx,
1173            exec_cmd_rx,
1174            signal_rx,
1175            signal_tx,
1176        );
1177        runner.bind_senders();
1178        get_system_command_sender()
1179            .send(test_system_command())
1180            .unwrap();
1181        get_system_event_sender().send(test_system_event()).unwrap();
1182        get_system_event_sender().send(test_system_event()).unwrap();
1183        let mut processed_by_channel = [0; 7];
1184        let mut processed_order = Vec::new();
1185
1186        let first = runner.poll_pending(|event| match event {
1187            PendingRunnerEvent::TimeEvent(_) => {
1188                processed_by_channel[0] += 1;
1189                processed_order.push("time");
1190            }
1191            PendingRunnerEvent::SystemEvent(_) => {
1192                processed_by_channel[1] += 1;
1193                processed_order.push("system_event");
1194            }
1195            PendingRunnerEvent::SystemCommand(_) => {
1196                processed_by_channel[2] += 1;
1197                processed_order.push("system_command");
1198            }
1199            PendingRunnerEvent::ExecEvent(_) => {
1200                processed_by_channel[3] += 1;
1201                processed_order.push("exec_event");
1202            }
1203            PendingRunnerEvent::ExecCommand(_) => {
1204                processed_by_channel[4] += 1;
1205                processed_order.push("exec_command");
1206            }
1207            PendingRunnerEvent::DataEvent(_) => {
1208                processed_by_channel[5] += 1;
1209                processed_order.push("data_event");
1210                data_evt_tx
1211                    .send((DataEvent::Data(Data::Quote(test_quote()))).into())
1212                    .unwrap();
1213            }
1214            PendingRunnerEvent::DataCommand(_) => {
1215                processed_by_channel[6] += 1;
1216                processed_order.push("data_command");
1217            }
1218        });
1219
1220        let second = runner.poll_pending(|event| match event {
1221            PendingRunnerEvent::DataEvent(_) => {
1222                processed_by_channel[5] += 1;
1223                processed_order.push("data_event");
1224            }
1225            _ => panic!("Unexpected runner event"),
1226        });
1227
1228        assert_eq!(first, 8);
1229        assert_eq!(second, 1);
1230        assert_eq!(processed_by_channel, [1, 2, 1, 1, 1, 2, 1]);
1231        assert_eq!(
1232            processed_order,
1233            [
1234                "time",
1235                "system_event",
1236                "system_event",
1237                "system_command",
1238                "exec_event",
1239                "exec_command",
1240                "data_event",
1241                "data_command",
1242                "data_event",
1243            ]
1244        );
1245    }
1246
1247    #[cfg(feature = "node")]
1248    #[tokio::test]
1249    async fn test_recv_processes_system_event_before_command() {
1250        let (_time_evt_tx, time_evt_rx) = tokio::sync::mpsc::unbounded_channel();
1251        let (_data_evt_tx, data_evt_rx) = tokio::sync::mpsc::unbounded_channel();
1252        let (_data_cmd_tx, data_cmd_rx) = tokio::sync::mpsc::unbounded_channel();
1253        let (_exec_evt_tx, exec_evt_rx) = tokio::sync::mpsc::unbounded_channel();
1254        let (_exec_cmd_tx, exec_cmd_rx) = tokio::sync::mpsc::unbounded_channel();
1255        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel();
1256        let mut runner = create_test_runner(
1257            time_evt_rx,
1258            data_evt_rx,
1259            data_cmd_rx,
1260            exec_evt_rx,
1261            exec_cmd_rx,
1262            signal_rx,
1263            signal_tx,
1264        );
1265
1266        runner
1267            .system_cmd_tx
1268            .send((test_system_command()).into())
1269            .unwrap();
1270        runner
1271            .system_evt_tx
1272            .send((test_system_event()).into())
1273            .unwrap();
1274
1275        assert!(matches!(
1276            runner.recv().await,
1277            Some(PendingRunnerEvent::SystemEvent(_))
1278        ));
1279        assert!(matches!(
1280            runner.recv().await,
1281            Some(PendingRunnerEvent::SystemCommand(_))
1282        ));
1283    }
1284
1285    #[cfg(feature = "node")]
1286    #[rstest]
1287    #[case::timers_and_data(true, true, false)]
1288    #[case::timers_only(true, false, false)]
1289    #[case::data_only(false, true, false)]
1290    #[case::execution_backlog(true, true, true)]
1291    #[tokio::test]
1292    async fn test_runner_dispatch_progress_under_load(
1293        #[case] timers: bool,
1294        #[case] data: bool,
1295        #[case] execution: bool,
1296        #[values(false, true)] receive: bool,
1297    ) {
1298        let mut runner = AsyncRunner::new();
1299        let stop = runner.signal_tx.clone();
1300        let callback = TimeEventCallback::RustLocal(Rc::new(move |event| {
1301            if event.ts_init == UnixNanos::from(8) {
1302                stop.send(()).unwrap();
1303            }
1304        }));
1305        let received = Rc::new(std::cell::Cell::new(0_u64));
1306        let observed = received.clone();
1307        let stop = runner.signal_tx.clone();
1308        let exec_event = || {
1309            ExecutionEvent::Order(OrderEventAny::Submitted(
1310                OrderSubmittedSpec::builder().build(),
1311            ))
1312        };
1313        let exec_tx = runner.exec_evt_tx.clone();
1314        msgbus::register_data_endpoint(
1315            MessagingSwitchboard::data_engine_process_data(),
1316            TypedIntoHandler::from(move |event| {
1317                let Data::Quote(quote) = event else {
1318                    panic!("Expected quote")
1319                };
1320                observed.set(observed.get() + 1);
1321                assert_eq!(quote.ts_init.as_u64(), observed.get());
1322                // Execution between data events must not restart the data rotation
1323                exec_tx.send(exec_event().into()).unwrap();
1324
1325                if !timers && quote.ts_init == UnixNanos::from(8) {
1326                    stop.send(()).unwrap();
1327                }
1328            }),
1329        );
1330        let command = DataCommand::Subscribe(SubscribeCommand::Data(SubscribeCustomData {
1331            client_id: Some(ClientId::from("TEST")),
1332            venue: None,
1333            data_type: DataType::new("QuoteTick", None, None),
1334            command_id: UUID4::new(),
1335            ts_init: UnixNanos::default(),
1336            correlation_id: None,
1337            params: None,
1338        }));
1339
1340        // Stop halfway through a ready backlog, before starvation can disappear by draining it
1341        for ts in (1..=16_u64).map(UnixNanos::from) {
1342            if timers {
1343                let event = TimeEvent::new("fairness".into(), UUID4::new(), ts, ts);
1344                runner
1345                    .time_evt_tx
1346                    .send(TimeEventMessage::new(event, callback.clone()).into())
1347                    .unwrap();
1348            }
1349
1350            if data {
1351                let mut quote = test_quote();
1352                quote.ts_init = ts;
1353                runner
1354                    .data_evt_tx
1355                    .send(DataEvent::Data(Data::Quote(quote)).into())
1356                    .unwrap();
1357            }
1358
1359            if execution {
1360                runner.exec_evt_tx.send(exec_event().into()).unwrap();
1361            }
1362            runner.data_cmd_tx.send(command.clone().into()).unwrap();
1363        }
1364
1365        if receive {
1366            while runner.signal_rx.try_recv().is_err() {
1367                match runner.recv().await.unwrap() {
1368                    PendingRunnerEvent::TimeEvent(event) => {
1369                        let _ = AsyncRunner::handle_time_event(event);
1370                    }
1371                    PendingRunnerEvent::DataEvent(event) => AsyncRunner::dispatch_data_event(event),
1372                    PendingRunnerEvent::DataCommand(cmd) => AsyncRunner::handle_data_command(cmd),
1373                    PendingRunnerEvent::ExecEvent(event) => AsyncRunner::dispatch_exec_event(event),
1374                    _ => panic!("Unexpected runner event"),
1375                }
1376            }
1377        } else {
1378            runner.run().await.unwrap();
1379        }
1380
1381        if execution {
1382            assert!((1..16).contains(&runner.channels.exec_evt_rx.len()));
1383            assert_eq!(received.get(), 0);
1384        } else {
1385            assert!((1..16).contains(&runner.channels.data_cmd_rx.len()));
1386
1387            if data {
1388                assert!(received.get() > 0);
1389                assert!(!runner.channels.data_evt_rx.is_empty());
1390            }
1391        }
1392
1393        if timers {
1394            assert_eq!(runner.channels.time_evt_rx.len(), 8);
1395        }
1396    }
1397
1398    #[rstest]
1399    fn test_async_data_command_sender_creation() {
1400        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
1401        let sender = AsyncDataCommandSender::new(tx);
1402        assert!(format!("{sender:?}").contains("AsyncDataCommandSender"));
1403    }
1404
1405    #[cfg(feature = "node")]
1406    #[rstest]
1407    fn test_runner_senders_shutdown_logging() {
1408        struct ErrorCapture(std::sync::Mutex<Vec<String>>);
1409
1410        impl log::Log for ErrorCapture {
1411            fn enabled(&self, metadata: &log::Metadata<'_>) -> bool {
1412                metadata.level() == log::Level::Error
1413                    && metadata.target() == "nautilus_live::runner"
1414            }
1415
1416            fn log(&self, record: &log::Record<'_>) {
1417                if self.enabled(record.metadata()) {
1418                    self.0.lock().unwrap().push(record.args().to_string());
1419                }
1420            }
1421
1422            fn flush(&self) {}
1423        }
1424
1425        static ERRORS: ErrorCapture = ErrorCapture(std::sync::Mutex::new(Vec::new()));
1426        log::set_logger(&ERRORS).unwrap();
1427        log::set_max_level(log::LevelFilter::Error);
1428
1429        let runner = AsyncRunner::new();
1430        let handle = LiveNodeHandle::new();
1431        handle.set_starting();
1432        runner.bind_senders_for_node(handle.clone());
1433        let sender = get_data_cmd_sender();
1434        let time_sender = get_time_event_sender();
1435        let mut channels = runner.take_channels();
1436
1437        let command = DataCommand::Subscribe(SubscribeCommand::Data(SubscribeCustomData {
1438            client_id: Some(ClientId::from("TEST")),
1439            venue: None,
1440            data_type: DataType::new("QuoteTick", None, None),
1441            command_id: UUID4::new(),
1442            ts_init: UnixNanos::default(),
1443            correlation_id: None,
1444            params: None,
1445        }));
1446
1447        let event = TimeEvent::new("shutdown-timer".into(), UUID4::new(), 17.into(), 23.into());
1448        let expected_events = vec![event.clone(); 2];
1449        let received = Rc::new(RefCell::new(Vec::new()));
1450        let observed = received.clone();
1451        let callback = TimeEventCallback::RustLocal(Rc::new(move |event| {
1452            observed.borrow_mut().push(event);
1453        }));
1454
1455        // Final draining still delivers commands and timer callbacks while receivers are alive
1456        for stopped in [false, true] {
1457            if stopped {
1458                handle.set_stopped();
1459            } else {
1460                handle.set_shutting_down();
1461            }
1462
1463            sender.execute(command.clone());
1464            assert_eq!(
1465                channels
1466                    .data_cmd_rx
1467                    .try_recv()
1468                    .unwrap()
1469                    .dispatch(|command| command),
1470                command
1471            );
1472
1473            time_sender.send(TimeEventMessage::new(event.clone(), callback.clone()));
1474            assert!(AsyncRunner::handle_time_event(
1475                channels.time_evt_rx.try_recv().unwrap()
1476            ));
1477        }
1478
1479        assert_eq!(*received.borrow(), expected_events);
1480
1481        drop(channels);
1482        sender.execute(command.clone());
1483        time_sender.send(TimeEventMessage::new(event.clone(), callback.clone()));
1484        assert_eq!(*ERRORS.0.lock().unwrap(), Vec::<String>::new());
1485
1486        // A stopped previous node must not hide an unexpected closure in its replacement
1487        let runner = AsyncRunner::new();
1488        let handle = LiveNodeHandle::new();
1489        runner.bind_senders_for_node(handle.clone());
1490        let sender = get_data_cmd_sender();
1491        let time_sender = get_time_event_sender();
1492        drop(runner);
1493
1494        for shutting_down in [false, true] {
1495            if shutting_down {
1496                handle.set_shutting_down();
1497            } else {
1498                handle.set_starting();
1499            }
1500
1501            sender.execute(command.clone());
1502            time_sender.send(TimeEventMessage::new(event.clone(), callback.clone()));
1503        }
1504
1505        let expected_errors = vec![
1506            "Failed to send data command: channel closed",
1507            "Failed to send time event message: channel closed",
1508        ];
1509        assert_eq!(*ERRORS.0.lock().unwrap(), expected_errors.repeat(2));
1510
1511        ERRORS.0.lock().unwrap().clear();
1512        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
1513        drop(rx);
1514        AsyncDataCommandSender::new(tx).execute(command);
1515        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
1516        drop(rx);
1517        AsyncTimeEventSender::new(tx).send(TimeEventMessage::new(event, callback));
1518        assert_eq!(*ERRORS.0.lock().unwrap(), expected_errors);
1519        assert_eq!(*received.borrow(), expected_events);
1520    }
1521
1522    #[rstest]
1523    fn test_async_time_event_sender_creation() {
1524        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
1525        let sender = AsyncTimeEventSender::new(tx);
1526        assert!(format!("{sender:?}").contains("AsyncTimeEventSender"));
1527    }
1528
1529    #[tokio::test]
1530    async fn test_async_data_command_sender_execute() {
1531        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
1532        let sender = AsyncDataCommandSender::new(tx);
1533
1534        let command = DataCommand::Subscribe(SubscribeCommand::Data(SubscribeCustomData {
1535            client_id: Some(ClientId::from("TEST")),
1536            venue: None,
1537            data_type: DataType::new("QuoteTick", None, None),
1538            command_id: UUID4::new(),
1539            ts_init: UnixNanos::default(),
1540            correlation_id: None,
1541            params: None,
1542        }));
1543
1544        sender.execute(command.clone());
1545
1546        let received = rx.recv().await.unwrap().dispatch(|command| command);
1547        match (received, command) {
1548            (
1549                DataCommand::Subscribe(SubscribeCommand::Data(r)),
1550                DataCommand::Subscribe(SubscribeCommand::Data(c)),
1551            ) => {
1552                assert_eq!(r.client_id, c.client_id);
1553                assert_eq!(r.data_type, c.data_type);
1554            }
1555            _ => panic!("Command mismatch"),
1556        }
1557    }
1558
1559    #[rstest]
1560    #[case(false)]
1561    #[case(true)]
1562    fn test_async_command_senders_capture_owner_root(#[case] rooted: bool) {
1563        let runner = AsyncRunner::new();
1564        runner.bind_senders();
1565        let mut channels = runner.take_channels();
1566
1567        let data = DataCommand::Subscribe(SubscribeCommand::Data(SubscribeCustomData {
1568            client_id: Some(ClientId::from("ROOT")),
1569            venue: None,
1570            data_type: DataType::new("QuoteTick", None, None),
1571            command_id: UUID4::new(),
1572            ts_init: UnixNanos::from(7),
1573            correlation_id: None,
1574            params: None,
1575        }));
1576
1577        let trading = TradingCommand::CancelAllOrders(CancelAllOrders::new(
1578            TraderId::from("TRADER-001"),
1579            None,
1580            StrategyId::from("ROOT-001"),
1581            InstrumentId::from("EUR/USD.SIM"),
1582            Some(OrderSide::Sell),
1583            UUID4::new(),
1584            UnixNanos::from(11),
1585            None,
1586            None,
1587        ));
1588        let expected_data = data.clone();
1589        let expected_trading = trading.clone();
1590        let trigger = trading.clone();
1591
1592        let send = move || {
1593            get_data_cmd_sender().execute(data.clone());
1594            get_trading_cmd_sender().execute(TradingCommandMessage::new(
1595                MessagingSwitchboard::exec_engine_execute(),
1596                trading.clone(),
1597            ));
1598        };
1599
1600        if rooted {
1601            msgbus::register_trading_command_endpoint(
1602                MessagingSwitchboard::risk_engine_execute(),
1603                TypedIntoHandler::from(move |_| send()),
1604            );
1605            SyncTradingCommandSender.execute(TradingCommandMessage::new(
1606                MessagingSwitchboard::risk_engine_execute(),
1607                trigger,
1608            ));
1609            drain_trading_cmd_queue();
1610        } else {
1611            send();
1612        }
1613
1614        let data = channels.data_cmd_rx.try_recv().unwrap();
1615        let trading = channels.exec_cmd_rx.try_recv().unwrap();
1616
1617        let results = thread::spawn(move || {
1618            let data = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
1619                data.dispatch(|command| command)
1620            }));
1621
1622            let trading = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
1623                trading.dispatch(|message| (message.endpoint(), message.command().clone()))
1624            }));
1625
1626            (data, trading)
1627        })
1628        .join()
1629        .unwrap();
1630
1631        if rooted {
1632            for error in [results.0.unwrap_err(), results.1.unwrap_err()] {
1633                let message = error.downcast_ref::<String>().unwrap();
1634                assert!(message.contains("command context dispatched outside its owner thread"));
1635            }
1636        } else {
1637            assert_eq!(results.0.unwrap(), expected_data);
1638            assert_eq!(
1639                results.1.unwrap(),
1640                (
1641                    MessagingSwitchboard::exec_engine_execute(),
1642                    expected_trading
1643                ),
1644            );
1645        }
1646
1647        assert!(channels.data_cmd_rx.is_empty());
1648        assert!(channels.exec_cmd_rx.is_empty());
1649    }
1650
1651    #[rstest]
1652    #[case(false)]
1653    #[case(true)]
1654    fn test_event_senders_capture_owner_root(
1655        #[case] rooted: bool,
1656        #[values(false, true)] dispatch: bool,
1657    ) {
1658        let runner = AsyncRunner::new();
1659        runner.bind_senders();
1660        let mut channels = runner.take_channels();
1661        let expected_quote = test_quote();
1662        let expected_order = OrderSubmittedSpec::builder()
1663            .client_order_id(ClientOrderId::from("EVENT-017"))
1664            .build();
1665        let order = expected_order.clone();
1666
1667        let send = move || {
1668            get_data_event_sender()
1669                .send(DataEvent::Data(Data::Quote(expected_quote)))
1670                .unwrap();
1671            get_exec_event_sender()
1672                .send(ExecutionEvent::Order(OrderEventAny::Submitted(
1673                    order.clone(),
1674                )))
1675                .unwrap();
1676        };
1677
1678        if rooted {
1679            msgbus::register_trading_command_endpoint(
1680                MessagingSwitchboard::risk_engine_execute(),
1681                TypedIntoHandler::from(move |_| send()),
1682            );
1683            SyncTradingCommandSender.execute(TradingCommandMessage::new(
1684                MessagingSwitchboard::risk_engine_execute(),
1685                TradingCommand::QueryAccount(QueryAccount::new(
1686                    "TRADER-001".into(),
1687                    None,
1688                    "SIM-001".into(),
1689                    UUID4::new(),
1690                    19.into(),
1691                    None,
1692                    None,
1693                )),
1694            ));
1695            drain_trading_cmd_queue();
1696        } else {
1697            send();
1698        }
1699
1700        if dispatch {
1701            msgbus::register_data_endpoint(
1702                MessagingSwitchboard::data_engine_process_data(),
1703                TypedIntoHandler::from(move |data| {
1704                    assert_eq!(data, Data::Quote(expected_quote));
1705                    get_data_event_sender().send(DataEvent::Data(data)).unwrap();
1706                }),
1707            );
1708
1709            let order = expected_order.clone();
1710            msgbus::register_order_event_endpoint(
1711                MessagingSwitchboard::exec_engine_process(),
1712                TypedIntoHandler::from(move |event| {
1713                    assert_eq!(event, OrderEventAny::Submitted(order.clone()));
1714                    get_exec_event_sender()
1715                        .send(ExecutionEvent::Order(event))
1716                        .unwrap();
1717                }),
1718            );
1719
1720            AsyncRunner::dispatch_data_event(channels.data_evt_rx.try_recv().unwrap());
1721            AsyncRunner::dispatch_exec_event(channels.exec_evt_rx.try_recv().unwrap());
1722        }
1723
1724        let data = channels.data_evt_rx.try_recv().unwrap();
1725        let exec = channels.exec_evt_rx.try_recv().unwrap();
1726
1727        let (data, exec) = thread::spawn(move || {
1728            (
1729                std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
1730                    data.dispatch(|event| event)
1731                })),
1732                std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
1733                    exec.dispatch(|event| event)
1734                })),
1735            )
1736        })
1737        .join()
1738        .unwrap();
1739
1740        if rooted || dispatch {
1741            for error in [data.unwrap_err(), exec.unwrap_err()] {
1742                assert!(
1743                    error
1744                        .downcast_ref::<String>()
1745                        .unwrap()
1746                        .contains("command context dispatched outside its owner thread")
1747                );
1748            }
1749        } else {
1750            let DataEvent::Data(Data::Quote(quote)) = data.unwrap() else {
1751                panic!("expected quote")
1752            };
1753
1754            let ExecutionEvent::Order(OrderEventAny::Submitted(order)) = exec.unwrap() else {
1755                panic!("expected submitted order")
1756            };
1757
1758            assert_eq!(quote, expected_quote);
1759            assert_eq!(order, expected_order);
1760        }
1761
1762        assert!(channels.data_evt_rx.is_empty());
1763        assert!(channels.exec_evt_rx.is_empty());
1764    }
1765
1766    #[rstest]
1767    fn test_system_and_time_senders_capture_owner_root(#[values(false, true)] rooted: bool) {
1768        let runner = AsyncRunner::new();
1769        runner.bind_senders();
1770        let mut channels = runner.take_channels();
1771        let event = TimeEvent::new("system-time".into(), UUID4::new(), 17.into(), 23.into());
1772        let observed = Rc::new(RefCell::new(Vec::new()));
1773        let received = observed.clone();
1774        let expected_event = event.clone();
1775
1776        let callback = TimeEventCallback::RustLocal(Rc::new(move |event| {
1777            received.borrow_mut().push(event);
1778            get_system_command_sender()
1779                .send(test_system_command())
1780                .unwrap();
1781        }));
1782
1783        let send = move || {
1784            get_system_event_sender().send(test_system_event()).unwrap();
1785            get_system_command_sender()
1786                .send(test_system_command())
1787                .unwrap();
1788            get_time_event_sender().send(TimeEventMessage::new(event.clone(), callback.clone()));
1789        };
1790
1791        if rooted {
1792            msgbus::register_trading_command_endpoint(
1793                MessagingSwitchboard::risk_engine_execute(),
1794                TypedIntoHandler::from(move |_| send()),
1795            );
1796            SyncTradingCommandSender.execute(TradingCommandMessage::new(
1797                MessagingSwitchboard::risk_engine_execute(),
1798                TradingCommand::QueryAccount(QueryAccount::new(
1799                    "TRADER-001".into(),
1800                    None,
1801                    "SIM-001".into(),
1802                    UUID4::new(),
1803                    31.into(),
1804                    None,
1805                    None,
1806                )),
1807            ));
1808            drain_trading_cmd_queue();
1809        } else {
1810            send();
1811        }
1812
1813        let system_event = channels.system_evt_rx.try_recv().unwrap();
1814        let system_command = channels.system_cmd_rx.try_recv().unwrap();
1815        let time_event = channels.time_evt_rx.try_recv().unwrap();
1816        assert_eq!(system_event.is_rooted(), rooted);
1817        assert_eq!(system_command.is_rooted(), rooted);
1818        assert_eq!(time_event.is_rooted(), rooted);
1819        assert!(AsyncRunner::handle_time_event(time_event));
1820        let child = channels.system_cmd_rx.try_recv().unwrap();
1821
1822        assert_eq!(system_event.dispatch(|event| event), test_system_event());
1823        assert_eq!(
1824            system_command.dispatch(|command| command),
1825            test_system_command()
1826        );
1827        assert_eq!(*observed.borrow(), vec![expected_event]);
1828        assert!(child.is_rooted());
1829        assert_eq!(child.dispatch(|command| command), test_system_command());
1830        assert!(channels.time_evt_rx.is_empty());
1831        assert!(channels.system_evt_rx.is_empty());
1832        assert!(channels.system_cmd_rx.is_empty());
1833    }
1834
1835    #[cfg(not(all(feature = "simulation", madsim)))]
1836    #[tokio::test]
1837    async fn test_live_timer_started_under_root_emits_independent_event() {
1838        let runner = AsyncRunner::new();
1839        runner.bind_senders();
1840        let mut channels = runner.take_channels();
1841        let now = get_atomic_clock_realtime().get_time_ns();
1842        let name = Ustr::from("ROOTED_TIMER");
1843        let owner = thread::current().id();
1844        let observed = Rc::new(RefCell::new(Vec::new()));
1845        let received = observed.clone();
1846
1847        let callback = TimeEventCallback::RustLocal(Rc::new(move |event| {
1848            received
1849                .borrow_mut()
1850                .push((event.name, event.ts_event, thread::current().id()));
1851        }));
1852
1853        let timer = Rc::new(RefCell::new(None));
1854        let started = timer.clone();
1855        msgbus::register_trading_command_endpoint(
1856            MessagingSwitchboard::risk_engine_execute(),
1857            TypedIntoHandler::from(move |_| {
1858                get_system_command_sender()
1859                    .send(test_system_command())
1860                    .unwrap();
1861
1862                let mut timer = LiveTimer::new(
1863                    name,
1864                    NonZeroU64::new(1_000_000).unwrap(),
1865                    now,
1866                    Some(now),
1867                    callback.clone(),
1868                    true,
1869                    Some(get_time_event_sender()),
1870                );
1871                timer.start();
1872                *started.borrow_mut() = Some(timer);
1873            }),
1874        );
1875
1876        SyncTradingCommandSender.execute(TradingCommandMessage::new(
1877            MessagingSwitchboard::risk_engine_execute(),
1878            TradingCommand::QueryAccount(QueryAccount::new(
1879                "TRADER-001".into(),
1880                None,
1881                "SIM-001".into(),
1882                UUID4::new(),
1883                31.into(),
1884                None,
1885                None,
1886            )),
1887        ));
1888        drain_trading_cmd_queue();
1889        let witness = channels.system_cmd_rx.try_recv().unwrap();
1890        let message = tokio::time::timeout(Duration::from_secs(2), channels.time_evt_rx.recv())
1891            .await
1892            .unwrap()
1893            .unwrap();
1894        timer.borrow_mut().take().unwrap().cancel();
1895
1896        assert!(witness.is_rooted());
1897        assert!(!message.is_rooted());
1898        assert!(observed.borrow().is_empty());
1899        assert!(AsyncRunner::handle_time_event(message));
1900        assert_eq!(*observed.borrow(), vec![(name, now, owner)]);
1901        assert!(channels.time_evt_rx.is_empty());
1902    }
1903
1904    #[tokio::test]
1905    async fn test_async_time_event_sender_send() {
1906        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
1907        let sender = AsyncTimeEventSender::new(tx);
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        let message = TimeEventMessage::new(event, callback);
1917
1918        sender.send(message);
1919
1920        assert!(rx.recv().await.is_some());
1921    }
1922
1923    #[tokio::test]
1924    async fn test_runner_shutdown_signal() {
1925        // Create runner with manual channels to avoid global state
1926        let (_data_tx, data_evt_rx) =
1927            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataEvent>>();
1928        let (_cmd_tx, data_cmd_rx) =
1929            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataCommand>>();
1930        let (_time_tx, time_evt_rx) =
1931            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TimeEventMessage>>();
1932        let (_exec_evt_tx, exec_evt_rx) =
1933            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<ExecutionEvent>>();
1934        let (_exec_cmd_tx, exec_cmd_rx) =
1935            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TradingCommandMessage>>();
1936        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1937
1938        let mut runner = create_test_runner(
1939            time_evt_rx,
1940            data_evt_rx,
1941            data_cmd_rx,
1942            exec_evt_rx,
1943            exec_cmd_rx,
1944            signal_rx,
1945            signal_tx.clone(),
1946        );
1947
1948        // Start runner
1949        let runner_handle = tokio::spawn(async move {
1950            runner.run().await.unwrap();
1951        });
1952
1953        // Send shutdown signal
1954        signal_tx.send(()).unwrap();
1955
1956        // Runner should stop quickly
1957        let result = tokio::time::timeout(Duration::from_millis(100), runner_handle).await;
1958        assert!(result.is_ok(), "Runner should stop on signal");
1959    }
1960
1961    #[tokio::test]
1962    async fn test_runner_closes_on_channel_drop() {
1963        let (data_tx, data_evt_rx) =
1964            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataEvent>>();
1965        let (_cmd_tx, data_cmd_rx) =
1966            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataCommand>>();
1967        let (_time_tx, time_evt_rx) =
1968            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TimeEventMessage>>();
1969        let (_exec_evt_tx, exec_evt_rx) =
1970            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<ExecutionEvent>>();
1971        let (_exec_cmd_tx, exec_cmd_rx) =
1972            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TradingCommandMessage>>();
1973        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
1974
1975        let mut runner = create_test_runner(
1976            time_evt_rx,
1977            data_evt_rx,
1978            data_cmd_rx,
1979            exec_evt_rx,
1980            exec_cmd_rx,
1981            signal_rx,
1982            signal_tx.clone(),
1983        );
1984
1985        // Start runner
1986        let runner_handle = tokio::spawn(async move {
1987            runner.run().await.unwrap();
1988        });
1989
1990        drop(data_tx);
1991
1992        // Yield to let runner enter event loop before stop signal
1993        tokio::task::yield_now().await;
1994        signal_tx.send(()).ok();
1995
1996        // Runner should stop when channels close or on signal
1997        let result = tokio::time::timeout(Duration::from_millis(200), runner_handle).await;
1998        assert!(
1999            result.is_ok(),
2000            "Runner should stop when channels close or on signal"
2001        );
2002    }
2003
2004    #[tokio::test]
2005    async fn test_concurrent_event_sending() {
2006        let (data_evt_tx, data_evt_rx) =
2007            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataEvent>>();
2008        let (_data_cmd_tx, data_cmd_rx) =
2009            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataCommand>>();
2010        let (_time_evt_tx, time_evt_rx) =
2011            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TimeEventMessage>>();
2012        let (_exec_evt_tx, exec_evt_rx) =
2013            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<ExecutionEvent>>();
2014        let (_exec_cmd_tx, exec_cmd_rx) =
2015            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TradingCommandMessage>>();
2016        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
2017
2018        // Setup runner
2019        let mut runner = create_test_runner(
2020            time_evt_rx,
2021            data_evt_rx,
2022            data_cmd_rx,
2023            exec_evt_rx,
2024            exec_cmd_rx,
2025            signal_rx,
2026            signal_tx.clone(),
2027        );
2028
2029        // Spawn multiple concurrent senders
2030        let mut handles = vec![];
2031
2032        for _ in 0..5 {
2033            let tx_clone = data_evt_tx.clone();
2034
2035            let handle = tokio::spawn(async move {
2036                for _ in 0..20 {
2037                    let quote = test_quote();
2038                    tx_clone
2039                        .send(DataEvent::Data(Data::Quote(quote)).into())
2040                        .unwrap();
2041                    tokio::task::yield_now().await;
2042                }
2043            });
2044
2045            handles.push(handle);
2046        }
2047
2048        // Start runner in background
2049        let runner_handle = tokio::spawn(async move {
2050            runner.run().await.unwrap();
2051        });
2052
2053        // Wait for all senders
2054        for handle in handles {
2055            handle.await.unwrap();
2056        }
2057
2058        // Yield to let runner enter event loop before stop signal
2059        tokio::task::yield_now().await;
2060        signal_tx.send(()).unwrap();
2061
2062        let _ = tokio::time::timeout(Duration::from_millis(200), runner_handle).await;
2063    }
2064
2065    #[rstest]
2066    #[case(10)]
2067    #[case(100)]
2068    #[case(1000)]
2069    fn test_channel_send_performance(#[case] count: usize) {
2070        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
2071        let quote = test_quote();
2072
2073        // Send events
2074        for _ in 0..count {
2075            tx.send(DataEvent::Data(Data::Quote(quote))).unwrap();
2076        }
2077
2078        // Verify all received
2079        let mut received = 0;
2080
2081        while rx.try_recv().is_ok() {
2082            received += 1;
2083        }
2084
2085        assert_eq!(received, count);
2086    }
2087
2088    #[rstest]
2089    fn test_async_trading_command_sender_creation() {
2090        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
2091        let sender = AsyncTradingCommandSender::new(tx);
2092        assert!(format!("{sender:?}").contains("AsyncTradingCommandSender"));
2093    }
2094
2095    #[rstest]
2096    fn test_async_trading_command_sender_preserves_target_endpoints() {
2097        std::thread::spawn(|| {
2098            msgbus::get_message_bus().borrow_mut().dispose();
2099            let (risk_handler, risk_saving_handler) =
2100                get_typed_into_message_saving_handler::<TradingCommand>(Some(Ustr::from(
2101                    "RiskEngine.execute",
2102                )));
2103            msgbus::register_trading_command_endpoint(
2104                MessagingSwitchboard::risk_engine_execute(),
2105                risk_handler,
2106            );
2107            let (exec_handler, exec_saving_handler) =
2108                get_typed_into_message_saving_handler::<TradingCommand>(Some(Ustr::from(
2109                    "ExecEngine.execute",
2110                )));
2111            msgbus::register_trading_command_endpoint(
2112                MessagingSwitchboard::exec_engine_execute(),
2113                exec_handler,
2114            );
2115
2116            let (tx, mut rx) =
2117                tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TradingCommandMessage>>();
2118            let sender = AsyncTradingCommandSender::new(tx);
2119            sender.execute(TradingCommandMessage::new(
2120                MessagingSwitchboard::risk_engine_execute(),
2121                TradingCommand::CancelAllOrders(CancelAllOrders::new(
2122                    TraderId::from("TRADER-001"),
2123                    None,
2124                    StrategyId::from("RISK-001"),
2125                    InstrumentId::from("EUR/USD.SIM"),
2126                    Some(OrderSide::Buy),
2127                    UUID4::new(),
2128                    UnixNanos::default(),
2129                    None,
2130                    None,
2131                )),
2132            ));
2133            sender.execute(TradingCommandMessage::new(
2134                MessagingSwitchboard::exec_engine_execute(),
2135                TradingCommand::CancelAllOrders(CancelAllOrders::new(
2136                    TraderId::from("TRADER-001"),
2137                    None,
2138                    StrategyId::from("EXEC-001"),
2139                    InstrumentId::from("EUR/USD.SIM"),
2140                    Some(OrderSide::Sell),
2141                    UUID4::new(),
2142                    UnixNanos::default(),
2143                    None,
2144                    None,
2145                )),
2146            ));
2147
2148            AsyncRunner::handle_trading_command(rx.try_recv().unwrap());
2149            AsyncRunner::handle_trading_command(rx.try_recv().unwrap());
2150
2151            let risk_commands = risk_saving_handler.get_messages();
2152            let exec_commands = exec_saving_handler.get_messages();
2153            assert!(rx.try_recv().is_err());
2154            assert_eq!(risk_commands.len(), 1);
2155            assert_eq!(
2156                risk_commands[0].strategy_id(),
2157                Some(StrategyId::from("RISK-001"))
2158            );
2159            assert_eq!(exec_commands.len(), 1);
2160            assert_eq!(
2161                exec_commands[0].strategy_id(),
2162                Some(StrategyId::from("EXEC-001"))
2163            );
2164        })
2165        .join()
2166        .unwrap();
2167    }
2168
2169    #[rstest]
2170    fn test_async_runner_preserves_deferred_follow_up_order() {
2171        std::thread::spawn(|| {
2172            msgbus::get_message_bus().borrow_mut().dispose();
2173            let clock = Rc::new(RefCell::new(VirtualClock::new()));
2174            let cache = Rc::new(RefCell::new(Cache::default()));
2175            let exec_engine = Rc::new(RefCell::new(ExecutionEngine::new(clock, cache, None)));
2176            ExecutionEngine::register_msgbus_handlers(&exec_engine);
2177            msgbus::register_trading_command_endpoint(
2178                MessagingSwitchboard::risk_engine_execute(),
2179                TypedIntoHandler::from(|command: TradingCommand| {
2180                    msgbus::send_trading_command(
2181                        MessagingSwitchboard::exec_engine_queue_execute(),
2182                        command,
2183                    );
2184                }),
2185            );
2186
2187            let (exec_handler, exec_saving_handler) =
2188                get_typed_into_message_saving_handler::<TradingCommand>(Some(Ustr::from(
2189                    "ExecEngine.execute",
2190                )));
2191            msgbus::register_trading_command_endpoint(
2192                MessagingSwitchboard::exec_engine_execute(),
2193                exec_handler,
2194            );
2195
2196            let (tx, mut rx) =
2197                tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TradingCommandMessage>>();
2198            let sender = Arc::new(AsyncTradingCommandSender::new(tx));
2199            replace_exec_cmd_sender(sender.clone());
2200            sender.execute(TradingCommandMessage::new(
2201                MessagingSwitchboard::risk_engine_execute(),
2202                TradingCommand::CancelAllOrders(CancelAllOrders::new(
2203                    TraderId::from("TRADER-001"),
2204                    None,
2205                    StrategyId::from("FIRST-001"),
2206                    InstrumentId::from("EUR/USD.SIM"),
2207                    Some(OrderSide::Buy),
2208                    UUID4::new(),
2209                    UnixNanos::default(),
2210                    None,
2211                    None,
2212                )),
2213            ));
2214            sender.execute(TradingCommandMessage::new(
2215                MessagingSwitchboard::exec_engine_execute(),
2216                TradingCommand::CancelAllOrders(CancelAllOrders::new(
2217                    TraderId::from("TRADER-001"),
2218                    None,
2219                    StrategyId::from("SECOND-001"),
2220                    InstrumentId::from("EUR/USD.SIM"),
2221                    Some(OrderSide::Sell),
2222                    UUID4::new(),
2223                    UnixNanos::default(),
2224                    None,
2225                    None,
2226                )),
2227            ));
2228
2229            AsyncRunner::handle_trading_command(rx.try_recv().unwrap());
2230            AsyncRunner::handle_trading_command(rx.try_recv().unwrap());
2231
2232            let commands = exec_saving_handler.get_messages();
2233            let strategy_ids = commands
2234                .iter()
2235                .map(TradingCommand::strategy_id)
2236                .collect::<Vec<_>>();
2237            assert!(rx.try_recv().is_err());
2238            assert_eq!(commands.len(), 2);
2239            assert_eq!(
2240                strategy_ids,
2241                vec![
2242                    Some(StrategyId::from("FIRST-001")),
2243                    Some(StrategyId::from("SECOND-001"))
2244                ]
2245            );
2246        })
2247        .join()
2248        .unwrap();
2249    }
2250
2251    #[rstest]
2252    fn test_async_runner_dispatches_deferred_exec_command_once() {
2253        std::thread::spawn(|| {
2254            msgbus::get_message_bus().borrow_mut().dispose();
2255            let clock = Rc::new(RefCell::new(VirtualClock::new()));
2256            let cache = Rc::new(RefCell::new(Cache::default()));
2257            let exec_engine = Rc::new(RefCell::new(ExecutionEngine::new(clock, cache, None)));
2258            ExecutionEngine::register_msgbus_handlers(&exec_engine);
2259
2260            let (tx, mut rx) =
2261                tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TradingCommandMessage>>();
2262            replace_exec_cmd_sender(Arc::new(AsyncTradingCommandSender::new(tx)));
2263            let command = TradingCommand::CancelAllOrders(CancelAllOrders::new(
2264                TraderId::from("TRADER-001"),
2265                None,
2266                StrategyId::from("EXEC-001"),
2267                InstrumentId::from("EUR/USD.SIM"),
2268                Some(OrderSide::Buy),
2269                UUID4::new(),
2270                UnixNanos::default(),
2271                None,
2272                None,
2273            ));
2274
2275            msgbus::send_trading_command(
2276                MessagingSwitchboard::exec_engine_queue_execute(),
2277                command,
2278            );
2279            assert_eq!(exec_engine.borrow().command_count(), 0);
2280
2281            AsyncRunner::handle_trading_command(rx.try_recv().unwrap());
2282
2283            assert!(rx.try_recv().is_err());
2284            assert_eq!(exec_engine.borrow().command_count(), 1);
2285        })
2286        .join()
2287        .unwrap();
2288    }
2289
2290    #[tokio::test]
2291    async fn test_runner_processes_trading_commands() {
2292        let (_data_evt_tx, data_evt_rx) =
2293            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataEvent>>();
2294        let (_data_cmd_tx, data_cmd_rx) =
2295            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataCommand>>();
2296        let (_time_evt_tx, time_evt_rx) =
2297            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TimeEventMessage>>();
2298        let (_exec_evt_tx, exec_evt_rx) =
2299            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<ExecutionEvent>>();
2300        let (exec_cmd_tx, exec_cmd_rx) =
2301            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TradingCommandMessage>>();
2302        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
2303
2304        let mut runner = create_test_runner(
2305            time_evt_rx,
2306            data_evt_rx,
2307            data_cmd_rx,
2308            exec_evt_rx,
2309            exec_cmd_rx,
2310            signal_rx,
2311            signal_tx.clone(),
2312        );
2313
2314        let runner_handle = tokio::spawn(async move {
2315            runner.run().await.unwrap();
2316        });
2317
2318        let command = TradingCommand::CancelAllOrders(CancelAllOrders::new(
2319            TraderId::from("TRADER-001"),
2320            None,
2321            StrategyId::from("S-001"),
2322            InstrumentId::from("EUR/USD.SIM"),
2323            Some(OrderSide::Buy),
2324            UUID4::new(),
2325            UnixNanos::default(),
2326            None,
2327            None, // correlation_id
2328        ));
2329        exec_cmd_tx
2330            .send(
2331                TradingCommandMessage::new(MessagingSwitchboard::exec_engine_execute(), command)
2332                    .into(),
2333            )
2334            .unwrap();
2335
2336        tokio::task::yield_now().await;
2337        signal_tx.send(()).unwrap();
2338
2339        let result = tokio::time::timeout(Duration::from_millis(100), runner_handle).await;
2340        assert!(result.is_ok(), "Runner should process command and stop");
2341    }
2342
2343    #[tokio::test]
2344    async fn test_runner_processes_multiple_trading_commands() {
2345        let (_data_evt_tx, data_evt_rx) =
2346            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataEvent>>();
2347        let (_data_cmd_tx, data_cmd_rx) =
2348            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataCommand>>();
2349        let (_time_evt_tx, time_evt_rx) =
2350            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TimeEventMessage>>();
2351        let (_exec_evt_tx, exec_evt_rx) =
2352            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<ExecutionEvent>>();
2353        let (exec_cmd_tx, exec_cmd_rx) =
2354            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TradingCommandMessage>>();
2355        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
2356
2357        let mut runner = create_test_runner(
2358            time_evt_rx,
2359            data_evt_rx,
2360            data_cmd_rx,
2361            exec_evt_rx,
2362            exec_cmd_rx,
2363            signal_rx,
2364            signal_tx.clone(),
2365        );
2366
2367        let runner_handle = tokio::spawn(async move {
2368            runner.run().await.unwrap();
2369        });
2370
2371        for i in 0..10 {
2372            let strategy_id = format!("S-{i:03}");
2373            let command = TradingCommand::CancelAllOrders(CancelAllOrders::new(
2374                TraderId::from("TRADER-001"),
2375                None,
2376                StrategyId::from(strategy_id.as_str()),
2377                InstrumentId::from("EUR/USD.SIM"),
2378                Some(OrderSide::Buy),
2379                UUID4::new(),
2380                UnixNanos::default(),
2381                None,
2382                None, // correlation_id
2383            ));
2384            exec_cmd_tx
2385                .send(
2386                    TradingCommandMessage::new(
2387                        MessagingSwitchboard::exec_engine_execute(),
2388                        command,
2389                    )
2390                    .into(),
2391                )
2392                .unwrap();
2393        }
2394
2395        tokio::task::yield_now().await;
2396        signal_tx.send(()).unwrap();
2397
2398        let result = tokio::time::timeout(Duration::from_millis(100), runner_handle).await;
2399        assert!(
2400            result.is_ok(),
2401            "Runner should process all commands and stop"
2402        );
2403    }
2404
2405    #[tokio::test]
2406    async fn test_execution_event_order_channel() {
2407        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
2408
2409        let event = OrderSubmittedSpec::builder()
2410            .client_order_id(ClientOrderId::from("O-001"))
2411            .build();
2412
2413        tx.send(ExecutionEvent::Order(OrderEventAny::Submitted(event)))
2414            .unwrap();
2415
2416        let received = rx.recv().await.unwrap();
2417        match received {
2418            ExecutionEvent::Order(OrderEventAny::Submitted(e)) => {
2419                assert_eq!(e.client_order_id(), ClientOrderId::from("O-001"));
2420            }
2421            _ => panic!("Expected OrderSubmitted event"),
2422        }
2423    }
2424
2425    #[tokio::test]
2426    async fn test_execution_report_order_status_channel() {
2427        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
2428
2429        let report = OrderStatusReport::new(
2430            AccountId::from("SIM-001"),
2431            InstrumentId::from("EUR/USD.SIM"),
2432            Some(ClientOrderId::from("O-001")),
2433            VenueOrderId::from("V-001"),
2434            OrderSide::Buy.into(),
2435            OrderType::Market,
2436            TimeInForce::Gtc,
2437            OrderStatus::Accepted,
2438            Quantity::from(100_000),
2439            Quantity::from(100_000),
2440            UnixNanos::from(1),
2441            UnixNanos::from(2),
2442            UnixNanos::from(3),
2443            None,
2444        );
2445
2446        tx.send(ExecutionEvent::Report(ExecutionReport::Order(Box::new(
2447            report,
2448        ))))
2449        .unwrap();
2450
2451        let received = rx.recv().await.unwrap();
2452        match received {
2453            ExecutionEvent::Report(ExecutionReport::Order(r)) => {
2454                assert_eq!(r.venue_order_id.as_str(), "V-001");
2455                assert_eq!(r.order_status, OrderStatus::Accepted);
2456            }
2457            _ => panic!("Expected OrderStatusReport"),
2458        }
2459    }
2460
2461    #[tokio::test]
2462    async fn test_execution_report_fill() {
2463        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
2464
2465        let report = FillReport::new(
2466            AccountId::from("SIM-001"),
2467            InstrumentId::from("EUR/USD.SIM"),
2468            VenueOrderId::from("V-001"),
2469            TradeId::from("T-001"),
2470            OrderSide::Buy,
2471            Quantity::from(100_000),
2472            Price::from("1.10000"),
2473            Money::from("10 USD"),
2474            LiquiditySide::Taker,
2475            Some(ClientOrderId::from("O-001")),
2476            None,
2477            UnixNanos::from(1),
2478            UnixNanos::from(2),
2479            None,
2480        );
2481
2482        tx.send(ExecutionEvent::Report(ExecutionReport::Fill(Box::new(
2483            report,
2484        ))))
2485        .unwrap();
2486
2487        let received = rx.recv().await.unwrap();
2488        match received {
2489            ExecutionEvent::Report(ExecutionReport::Fill(r)) => {
2490                assert_eq!(r.venue_order_id.as_str(), "V-001");
2491                assert_eq!(r.trade_id.to_string(), "T-001");
2492            }
2493            _ => panic!("Expected FillReport"),
2494        }
2495    }
2496
2497    #[tokio::test]
2498    async fn test_execution_report_position() {
2499        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
2500
2501        let report = PositionStatusReport::new(
2502            AccountId::from("SIM-001"),
2503            InstrumentId::from("EUR/USD.SIM"),
2504            PositionSide::Long,
2505            Quantity::from(100_000),
2506            UnixNanos::from(1),
2507            UnixNanos::from(2),
2508            None,
2509            Some(PositionId::from("P-001")),
2510            None,
2511        );
2512
2513        tx.send(ExecutionEvent::Report(ExecutionReport::Position(Box::new(
2514            report,
2515        ))))
2516        .unwrap();
2517
2518        let received = rx.recv().await.unwrap();
2519        match received {
2520            ExecutionEvent::Report(ExecutionReport::Position(r)) => {
2521                assert_eq!(r.venue_position_id.unwrap().as_str(), "P-001");
2522            }
2523            _ => panic!("Expected PositionStatusReport"),
2524        }
2525    }
2526
2527    #[tokio::test]
2528    async fn test_execution_event_account() {
2529        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
2530
2531        let account_state = AccountState::new(
2532            AccountId::from("SIM-001"),
2533            AccountType::Cash,
2534            vec![],
2535            vec![],
2536            true,
2537            UUID4::new(),
2538            UnixNanos::from(1),
2539            UnixNanos::from(2),
2540            None,
2541        );
2542
2543        tx.send(ExecutionEvent::Account(account_state)).unwrap();
2544
2545        let received = rx.recv().await.unwrap();
2546        match received {
2547            ExecutionEvent::Account(r) => {
2548                assert_eq!(r.account_id.as_str(), "SIM-001");
2549            }
2550            _ => panic!("Expected AccountState"),
2551        }
2552    }
2553
2554    #[cfg(feature = "node")]
2555    #[rstest]
2556    #[case::cash(AccountType::Cash)]
2557    #[case::margin(AccountType::Margin)]
2558    fn test_account_event_publishes_after_portfolio_application(#[case] account_type: AccountType) {
2559        use nautilus_common::msgbus::{MessageBus, TypedHandler};
2560        use nautilus_model::{
2561            accounts::Account,
2562            types::{AccountBalance, Currency},
2563        };
2564        use nautilus_portfolio::Portfolio;
2565
2566        // Native logging tests use nextest's per-test process isolation.
2567        struct AccountLogCapture(std::sync::Mutex<Vec<String>>);
2568
2569        impl log::Log for AccountLogCapture {
2570            fn enabled(&self, metadata: &log::Metadata<'_>) -> bool {
2571                metadata.level() == log::Level::Info
2572                    && metadata.target() == "nautilus_portfolio::portfolio"
2573            }
2574
2575            fn log(&self, record: &log::Record<'_>) {
2576                if self.enabled(record.metadata()) {
2577                    self.0.lock().unwrap().push(record.args().to_string());
2578                }
2579            }
2580
2581            fn flush(&self) {}
2582        }
2583
2584        static LOGS: AccountLogCapture = AccountLogCapture(std::sync::Mutex::new(Vec::new()));
2585        log::set_logger(&LOGS).unwrap();
2586        log::set_max_level(log::LevelFilter::Info);
2587
2588        msgbus::set_message_bus(Rc::new(RefCell::new(MessageBus::new(
2589            TraderId::from("TESTER-001"),
2590            UUID4::new(),
2591            None,
2592            None,
2593        ))));
2594        let cache = Rc::new(RefCell::new(Cache::default()));
2595        let clock = Rc::new(RefCell::new(VirtualClock::new()));
2596        let _portfolio = Portfolio::new(clock, cache.clone(), None);
2597        let observed = Rc::new(RefCell::new(Vec::new()));
2598        let received = observed.clone();
2599        let source = cache.clone();
2600        let handler = TypedHandler::from(move |event: &AccountState| {
2601            let cache = source.borrow();
2602            let account = cache.account(&event.account_id).unwrap();
2603            assert_eq!(account.last_event().unwrap().event_id, event.event_id);
2604            assert_eq!(account.account_type(), event.account_type);
2605            assert_eq!(
2606                account.balances().into_values().collect::<Vec<_>>(),
2607                event.balances
2608            );
2609            received
2610                .borrow_mut()
2611                .push((event.event_id, account.events().len()));
2612        });
2613
2614        // Subscribers must see applied state regardless of their topic priority
2615        msgbus::subscribe_account_state("events.account.*".into(), handler.clone(), Some(20));
2616        let scoped = Rc::new(RefCell::new(Vec::new()));
2617        let scoped_received = scoped.clone();
2618        let scoped_handler = TypedHandler::from(move |event: &AccountState| {
2619            scoped_received.borrow_mut().push(event.event_id);
2620        });
2621        msgbus::subscribe_account_state(
2622            "events.account.SIM-001".into(),
2623            scoped_handler.clone(),
2624            None,
2625        );
2626        let mut expected = Vec::new();
2627        let mut expected_scoped = Vec::new();
2628        for (index, (account_id, total, locked, free, history_len)) in [
2629            ("SIM-001", "100 USD", "20 USD", "80 USD", 1),
2630            ("SIM-002", "200 USD", "50 USD", "150 USD", 1),
2631            ("SIM-001", "90 USD", "10 USD", "80 USD", 2),
2632            ("SIM-001", "0 USD", "0 USD", "0 USD", 3),
2633        ]
2634        .into_iter()
2635        .enumerate()
2636        {
2637            let event = AccountState::new(
2638                AccountId::from(account_id),
2639                account_type,
2640                vec![AccountBalance::new(
2641                    Money::from(total),
2642                    Money::from(locked),
2643                    Money::from(free),
2644                )],
2645                vec![],
2646                true,
2647                UUID4::new(),
2648                UnixNanos::from(index as u64 + 1),
2649                UnixNanos::from(index as u64 + 1),
2650                Some(Currency::USD()),
2651            );
2652            expected.push((event.event_id, history_len));
2653            if account_id == "SIM-001" {
2654                expected_scoped.push(event.event_id);
2655            }
2656            let expected_log = format!("Updated {event}");
2657            AsyncRunner::handle_exec_event(ExecutionEvent::Account(event));
2658            assert_eq!(
2659                LOGS.0
2660                    .lock()
2661                    .unwrap()
2662                    .iter()
2663                    .filter(|line| **line == expected_log)
2664                    .count(),
2665                1
2666            );
2667        }
2668        assert_eq!(*observed.borrow(), expected);
2669        assert_eq!(*scoped.borrow(), expected_scoped);
2670        let cache = cache.borrow();
2671        assert_eq!(
2672            cache
2673                .account(&AccountId::from("SIM-001"))
2674                .unwrap()
2675                .events()
2676                .len(),
2677            3
2678        );
2679        assert_eq!(
2680            cache
2681                .account(&AccountId::from("SIM-002"))
2682                .unwrap()
2683                .events()
2684                .len(),
2685            1
2686        );
2687        msgbus::unsubscribe_account_state("events.account.*".into(), &handler);
2688        msgbus::unsubscribe_account_state("events.account.SIM-001".into(), &scoped_handler);
2689    }
2690
2691    #[cfg(feature = "node")]
2692    #[rstest]
2693    #[case::cash_update(AccountType::Cash, true)]
2694    #[case::wallet_initial(AccountType::Wallet, false)]
2695    #[case::wallet_update(AccountType::Wallet, true)]
2696    fn test_rejected_account_event_is_not_published(
2697        #[case] account_type: AccountType,
2698        #[case] existing_account: bool,
2699    ) {
2700        use nautilus_common::msgbus::{MessageBus, TypedHandler};
2701        use nautilus_model::types::{AccountBalance, Currency};
2702        use nautilus_portfolio::Portfolio;
2703
2704        msgbus::set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
2705        let cache = Rc::new(RefCell::new(Cache::default()));
2706        let mut portfolio = Portfolio::new(
2707            Rc::new(RefCell::new(VirtualClock::new())),
2708            cache.clone(),
2709            None,
2710        );
2711        let state = |amount: &str| {
2712            AccountState::new(
2713                AccountId::from("SIM-001"),
2714                account_type,
2715                vec![AccountBalance::new(
2716                    Money::from(amount),
2717                    Money::zero(Currency::USD()),
2718                    Money::from(amount),
2719                )],
2720                vec![],
2721                true,
2722                UUID4::new(),
2723                UnixNanos::from(1),
2724                UnixNanos::from(1),
2725                (account_type != AccountType::Wallet).then_some(Currency::USD()),
2726            )
2727        };
2728        let initial = state("100.00 USD");
2729
2730        if existing_account {
2731            portfolio.update_account(&initial);
2732        }
2733        let received = Rc::new(RefCell::new(Vec::new()));
2734        let handler = TypedHandler::from({
2735            let received = received.clone();
2736            move |event: &AccountState| received.borrow_mut().push(event.clone())
2737        });
2738        msgbus::subscribe_account_state("events.account.*".into(), handler.clone(), None);
2739        AsyncRunner::handle_exec_event(ExecutionEvent::Account(state("-1.00 USD")));
2740        assert!(received.borrow().is_empty());
2741        let cache = cache.borrow();
2742
2743        if existing_account {
2744            assert_eq!(
2745                cache.account(&initial.account_id).unwrap().last_event(),
2746                Some(initial)
2747            );
2748        } else {
2749            assert!(cache.account(&initial.account_id).is_none());
2750        }
2751        msgbus::unsubscribe_account_state("events.account.*".into(), &handler);
2752    }
2753
2754    #[tokio::test]
2755    async fn test_runner_stop_method() {
2756        let (_data_tx, data_evt_rx) =
2757            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataEvent>>();
2758        let (_cmd_tx, data_cmd_rx) =
2759            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataCommand>>();
2760        let (_time_tx, time_evt_rx) =
2761            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TimeEventMessage>>();
2762        let (_exec_evt_tx, exec_evt_rx) =
2763            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<ExecutionEvent>>();
2764        let (_exec_cmd_tx, exec_cmd_rx) =
2765            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TradingCommandMessage>>();
2766        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
2767
2768        let mut runner = create_test_runner(
2769            time_evt_rx,
2770            data_evt_rx,
2771            data_cmd_rx,
2772            exec_evt_rx,
2773            exec_cmd_rx,
2774            signal_rx,
2775            signal_tx.clone(),
2776        );
2777
2778        let runner_handle = tokio::spawn(async move {
2779            runner.run().await.unwrap();
2780        });
2781
2782        // Use stop via signal_tx directly
2783        signal_tx.send(()).unwrap();
2784
2785        let result = tokio::time::timeout(Duration::from_millis(100), runner_handle).await;
2786        assert!(result.is_ok(), "Runner should stop when stop() is called");
2787    }
2788
2789    #[tokio::test]
2790    async fn test_all_event_types_integration() {
2791        let (data_evt_tx, data_evt_rx) =
2792            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataEvent>>();
2793        let (data_cmd_tx, data_cmd_rx) =
2794            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataCommand>>();
2795        let (time_evt_tx, time_evt_rx) =
2796            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TimeEventMessage>>();
2797        let (exec_evt_tx, exec_evt_rx) =
2798            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<ExecutionEvent>>();
2799        let (_exec_cmd_tx, exec_cmd_rx) =
2800            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TradingCommandMessage>>();
2801        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
2802
2803        let mut runner = create_test_runner(
2804            time_evt_rx,
2805            data_evt_rx,
2806            data_cmd_rx,
2807            exec_evt_rx,
2808            exec_cmd_rx,
2809            signal_rx,
2810            signal_tx.clone(),
2811        );
2812
2813        let runner_handle = tokio::spawn(async move {
2814            runner.run().await.unwrap();
2815        });
2816
2817        // Send data event
2818        let quote = test_quote();
2819        data_evt_tx
2820            .send((DataEvent::Data(Data::Quote(quote))).into())
2821            .unwrap();
2822
2823        // Send data command
2824        let command = DataCommand::Subscribe(SubscribeCommand::Data(SubscribeCustomData {
2825            client_id: Some(ClientId::from("TEST")),
2826            venue: None,
2827            data_type: DataType::new("QuoteTick", None, None),
2828            command_id: UUID4::new(),
2829            ts_init: UnixNanos::default(),
2830            correlation_id: None,
2831            params: None,
2832        }));
2833
2834        data_cmd_tx.send(command.into()).unwrap();
2835
2836        // Send time event
2837        let event = TimeEvent::new(
2838            Ustr::from("test"),
2839            UUID4::new(),
2840            UnixNanos::from(1),
2841            UnixNanos::from(2),
2842        );
2843        let callback = TimeEventCallback::from(|_: TimeEvent| {});
2844        let message = TimeEventMessage::new(event, callback);
2845        time_evt_tx.send((message).into()).unwrap();
2846
2847        // Send execution order event
2848        let order_event = OrderSubmittedSpec::builder()
2849            .client_order_id(ClientOrderId::from("O-001"))
2850            .build();
2851        exec_evt_tx
2852            .send((ExecutionEvent::Order(OrderEventAny::Submitted(order_event))).into())
2853            .unwrap();
2854
2855        // Send execution report (OrderStatus)
2856        let order_status = OrderStatusReport::new(
2857            AccountId::from("SIM-001"),
2858            InstrumentId::from("EUR/USD.SIM"),
2859            Some(ClientOrderId::from("O-001")),
2860            VenueOrderId::from("V-001"),
2861            OrderSide::Buy.into(),
2862            OrderType::Market,
2863            TimeInForce::Gtc,
2864            OrderStatus::Accepted,
2865            Quantity::from(100_000),
2866            Quantity::from(100_000),
2867            UnixNanos::from(1),
2868            UnixNanos::from(2),
2869            UnixNanos::from(3),
2870            None,
2871        );
2872        exec_evt_tx
2873            .send((ExecutionEvent::Report(ExecutionReport::Order(Box::new(order_status)))).into())
2874            .unwrap();
2875
2876        // Send execution report (Fill)
2877        let fill = FillReport::new(
2878            AccountId::from("SIM-001"),
2879            InstrumentId::from("EUR/USD.SIM"),
2880            VenueOrderId::from("V-001"),
2881            TradeId::from("T-001"),
2882            OrderSide::Buy,
2883            Quantity::from(100_000),
2884            Price::from("1.10000"),
2885            Money::from("10 USD"),
2886            LiquiditySide::Taker,
2887            Some(ClientOrderId::from("O-001")),
2888            None,
2889            UnixNanos::from(1),
2890            UnixNanos::from(2),
2891            None,
2892        );
2893        exec_evt_tx
2894            .send((ExecutionEvent::Report(ExecutionReport::Fill(Box::new(fill)))).into())
2895            .unwrap();
2896
2897        // Send execution report (Position)
2898        let position = PositionStatusReport::new(
2899            AccountId::from("SIM-001"),
2900            InstrumentId::from("EUR/USD.SIM"),
2901            PositionSide::Long,
2902            Quantity::from(100_000),
2903            UnixNanos::from(1),
2904            UnixNanos::from(2),
2905            None,
2906            Some(PositionId::from("P-001")),
2907            None,
2908        );
2909        exec_evt_tx
2910            .send((ExecutionEvent::Report(ExecutionReport::Position(Box::new(position)))).into())
2911            .unwrap();
2912
2913        // Send account event
2914        let account_state = AccountState::new(
2915            AccountId::from("SIM-001"),
2916            AccountType::Cash,
2917            vec![],
2918            vec![],
2919            true,
2920            UUID4::new(),
2921            UnixNanos::from(1),
2922            UnixNanos::from(2),
2923            None,
2924        );
2925        exec_evt_tx
2926            .send((ExecutionEvent::Account(account_state)).into())
2927            .unwrap();
2928
2929        // Yield to let runner enter event loop before stop signal
2930        tokio::task::yield_now().await;
2931        signal_tx.send(()).unwrap();
2932
2933        let result = tokio::time::timeout(Duration::from_millis(200), runner_handle).await;
2934        assert!(
2935            result.is_ok(),
2936            "Runner should process all event types and stop cleanly"
2937        );
2938    }
2939
2940    #[tokio::test]
2941    async fn test_runner_handle_stops_runner() {
2942        let (_data_tx, data_evt_rx) =
2943            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataEvent>>();
2944        let (_cmd_tx, data_cmd_rx) =
2945            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataCommand>>();
2946        let (_time_tx, time_evt_rx) =
2947            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TimeEventMessage>>();
2948        let (_exec_evt_tx, exec_evt_rx) =
2949            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<ExecutionEvent>>();
2950        let (_exec_cmd_tx, exec_cmd_rx) =
2951            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TradingCommandMessage>>();
2952        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
2953
2954        let mut runner = create_test_runner(
2955            time_evt_rx,
2956            data_evt_rx,
2957            data_cmd_rx,
2958            exec_evt_rx,
2959            exec_cmd_rx,
2960            signal_rx,
2961            signal_tx.clone(),
2962        );
2963
2964        // Get handle before moving runner
2965        let handle = runner.handle();
2966
2967        let runner_handle = tokio::spawn(async move {
2968            runner.run().await.unwrap();
2969        });
2970
2971        // Use handle to stop
2972        handle.stop();
2973
2974        let result = tokio::time::timeout(Duration::from_millis(100), runner_handle).await;
2975        assert!(result.is_ok(), "Runner should stop via handle");
2976    }
2977
2978    #[tokio::test]
2979    async fn test_runner_handle_is_cloneable() {
2980        let (signal_tx, _signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
2981        let handle = AsyncRunnerHandle { signal_tx };
2982
2983        let handle2 = handle.clone();
2984
2985        // Both handles should be able to send stop signals
2986        assert!(handle.signal_tx.send(()).is_ok());
2987        assert!(handle2.signal_tx.send(()).is_ok());
2988    }
2989
2990    #[tokio::test]
2991    async fn test_runner_processes_events_before_stop() {
2992        let (data_evt_tx, data_evt_rx) =
2993            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataEvent>>();
2994        let (_cmd_tx, data_cmd_rx) =
2995            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<DataCommand>>();
2996        let (_time_tx, time_evt_rx) =
2997            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TimeEventMessage>>();
2998        let (_exec_evt_tx, exec_evt_rx) =
2999            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<ExecutionEvent>>();
3000        let (_exec_cmd_tx, exec_cmd_rx) =
3001            tokio::sync::mpsc::unbounded_channel::<DispatchMessage<TradingCommandMessage>>();
3002        let (signal_tx, signal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
3003
3004        let mut runner = create_test_runner(
3005            time_evt_rx,
3006            data_evt_rx,
3007            data_cmd_rx,
3008            exec_evt_rx,
3009            exec_cmd_rx,
3010            signal_rx,
3011            signal_tx.clone(),
3012        );
3013
3014        let handle = runner.handle();
3015
3016        // Send events before starting runner
3017        for _ in 0..10 {
3018            let quote = test_quote();
3019            data_evt_tx
3020                .send((DataEvent::Data(Data::Quote(quote))).into())
3021                .unwrap();
3022        }
3023
3024        let runner_handle = tokio::spawn(async move {
3025            runner.run().await.unwrap();
3026        });
3027
3028        // Yield to let runner enter event loop before stop signal
3029        tokio::task::yield_now().await;
3030        handle.stop();
3031
3032        let result = tokio::time::timeout(Duration::from_millis(200), runner_handle).await;
3033        assert!(result.is_ok(), "Runner should process events and stop");
3034    }
3035
3036    #[rstest]
3037    fn test_new_does_not_bind_tls() {
3038        std::thread::spawn(|| {
3039            let _runner = AsyncRunner::new();
3040            assert!(try_get_time_event_sender().is_none());
3041            assert!(try_get_system_command_sender().is_none());
3042            assert!(try_get_system_event_sender().is_none());
3043            assert!(try_get_trading_cmd_sender().is_none());
3044        })
3045        .join()
3046        .unwrap();
3047    }
3048
3049    #[rstest]
3050    fn test_bind_senders_routes_to_runner_channels() {
3051        std::thread::spawn(|| {
3052            let mut runner = AsyncRunner::new();
3053            runner.bind_senders();
3054
3055            get_data_cmd_sender().execute(DataCommand::Subscribe(SubscribeCommand::Data(
3056                SubscribeCustomData {
3057                    client_id: Some(ClientId::from("TEST")),
3058                    venue: None,
3059                    data_type: DataType::new("test", None, None),
3060                    command_id: UUID4::new(),
3061                    ts_init: UnixNanos::default(),
3062                    correlation_id: None,
3063                    params: None,
3064                },
3065            )));
3066
3067            assert!(runner.channels.data_cmd_rx.try_recv().is_ok());
3068
3069            get_trading_cmd_sender().execute(TradingCommandMessage::new(
3070                MessagingSwitchboard::exec_engine_execute(),
3071                TradingCommand::CancelAllOrders(CancelAllOrders::new(
3072                    TraderId::from("TRADER-001"),
3073                    None,
3074                    StrategyId::from("S-001"),
3075                    InstrumentId::from("EUR/USD.SIM"),
3076                    Some(OrderSide::Buy),
3077                    UUID4::new(),
3078                    UnixNanos::default(),
3079                    None,
3080                    None, // correlation_id
3081                )),
3082            ));
3083            assert!(runner.channels.exec_cmd_rx.try_recv().is_ok());
3084
3085            let event = TimeEvent::new(
3086                Ustr::from("test"),
3087                UUID4::new(),
3088                UnixNanos::from(1),
3089                UnixNanos::from(2),
3090            );
3091            let callback = TimeEventCallback::from(|_: TimeEvent| {});
3092            get_time_event_sender().send(TimeEventMessage::new(event, callback));
3093            assert!(runner.channels.time_evt_rx.try_recv().is_ok());
3094
3095            get_system_event_sender().send(test_system_event()).unwrap();
3096            assert_eq!(
3097                runner
3098                    .channels
3099                    .system_evt_rx
3100                    .try_recv()
3101                    .unwrap()
3102                    .dispatch(|value| value),
3103                test_system_event()
3104            );
3105
3106            get_system_command_sender()
3107                .send(test_system_command())
3108                .unwrap();
3109            assert_eq!(
3110                runner
3111                    .channels
3112                    .system_cmd_rx
3113                    .try_recv()
3114                    .unwrap()
3115                    .dispatch(|value| value),
3116                test_system_command()
3117            );
3118
3119            get_data_event_sender()
3120                .send(DataEvent::Data(Data::Quote(test_quote())))
3121                .unwrap();
3122            assert!(runner.channels.data_evt_rx.try_recv().is_ok());
3123
3124            let account = AccountState::new(
3125                AccountId::from("SIM-001"),
3126                AccountType::Cash,
3127                vec![],
3128                vec![],
3129                true,
3130                UUID4::new(),
3131                UnixNanos::from(1),
3132                UnixNanos::from(2),
3133                None,
3134            );
3135            get_exec_event_sender()
3136                .send(ExecutionEvent::Account(account))
3137                .unwrap();
3138            assert!(runner.channels.exec_evt_rx.try_recv().is_ok());
3139        })
3140        .join()
3141        .unwrap();
3142    }
3143
3144    #[cfg(feature = "node")]
3145    #[rstest]
3146    fn test_drain_pending_system_events_keeps_data_events_separate() {
3147        std::thread::spawn(|| {
3148            let mut runner = AsyncRunner::new();
3149            runner.bind_senders();
3150            let system_event = test_system_event();
3151
3152            get_system_event_sender().send(system_event).unwrap();
3153            get_data_event_sender()
3154                .send(DataEvent::Data(Data::Quote(test_quote())))
3155                .unwrap();
3156
3157            let system_events = runner
3158                .drain_pending_system_events()
3159                .into_iter()
3160                .map(|message| message.dispatch(|value| value))
3161                .collect::<Vec<_>>();
3162
3163            assert_eq!(system_events, vec![system_event]);
3164            assert!(runner.channels.system_evt_rx.try_recv().is_err());
3165            assert!(runner.channels.data_evt_rx.try_recv().is_ok());
3166        })
3167        .join()
3168        .unwrap();
3169    }
3170
3171    #[cfg(feature = "node")]
3172    #[rstest]
3173    fn test_drain_pending_system_commands_keeps_events_separate() {
3174        std::thread::spawn(|| {
3175            let mut runner = AsyncRunner::new();
3176            runner.bind_senders();
3177            let system_command = test_system_command();
3178
3179            get_system_command_sender().send(system_command).unwrap();
3180            get_system_event_sender().send(test_system_event()).unwrap();
3181
3182            let system_commands = runner
3183                .drain_pending_system_commands()
3184                .into_iter()
3185                .map(|message| message.dispatch(|value| value))
3186                .collect::<Vec<_>>();
3187
3188            assert_eq!(system_commands, vec![system_command]);
3189            assert!(runner.channels.system_cmd_rx.try_recv().is_err());
3190            assert!(runner.channels.system_evt_rx.try_recv().is_ok());
3191        })
3192        .join()
3193        .unwrap();
3194    }
3195
3196    #[rstest]
3197    fn test_bind_senders_reclaims_tls_from_previous_runner() {
3198        std::thread::spawn(|| {
3199            let mut runner1 = AsyncRunner::new();
3200            runner1.bind_senders();
3201
3202            let mut runner2 = AsyncRunner::new();
3203            runner2.bind_senders();
3204
3205            get_data_cmd_sender().execute(DataCommand::Subscribe(SubscribeCommand::Data(
3206                SubscribeCustomData {
3207                    client_id: Some(ClientId::from("TEST")),
3208                    venue: None,
3209                    data_type: DataType::new("test", None, None),
3210                    command_id: UUID4::new(),
3211                    ts_init: UnixNanos::default(),
3212                    correlation_id: None,
3213                    params: None,
3214                },
3215            )));
3216
3217            assert!(runner2.channels.data_cmd_rx.try_recv().is_ok());
3218            assert!(runner1.channels.data_cmd_rx.try_recv().is_err());
3219        })
3220        .join()
3221        .unwrap();
3222    }
3223
3224    #[tokio::test]
3225    async fn test_execution_event_order_submitted_batch_channel() {
3226        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
3227
3228        let events = vec![
3229            OrderSubmittedSpec::builder()
3230                .client_order_id(ClientOrderId::from("O-001"))
3231                .build(),
3232            OrderSubmittedSpec::builder()
3233                .client_order_id(ClientOrderId::from("O-002"))
3234                .build(),
3235        ];
3236
3237        let batch = OrderSubmittedBatch::new(events);
3238        tx.send(ExecutionEvent::OrderSubmittedBatch(batch)).unwrap();
3239
3240        let received = rx.recv().await.unwrap();
3241        match received {
3242            ExecutionEvent::OrderSubmittedBatch(b) => {
3243                assert_eq!(b.len(), 2);
3244                assert_eq!(b.events[0].client_order_id, ClientOrderId::from("O-001"));
3245                assert_eq!(b.events[1].client_order_id, ClientOrderId::from("O-002"));
3246            }
3247            _ => panic!("Expected OrderSubmittedBatch event"),
3248        }
3249    }
3250
3251    #[tokio::test]
3252    async fn test_execution_event_order_accepted_batch_channel() {
3253        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
3254
3255        let events = vec![
3256            OrderAcceptedSpec::builder()
3257                .client_order_id(ClientOrderId::from("O-001"))
3258                .build(),
3259            OrderAcceptedSpec::builder()
3260                .client_order_id(ClientOrderId::from("O-002"))
3261                .build(),
3262        ];
3263
3264        let batch = OrderAcceptedBatch::new(events);
3265        tx.send(ExecutionEvent::OrderAcceptedBatch(batch)).unwrap();
3266
3267        let received = rx.recv().await.unwrap();
3268        match received {
3269            ExecutionEvent::OrderAcceptedBatch(b) => {
3270                assert_eq!(b.len(), 2);
3271                assert_eq!(b.events[0].client_order_id, ClientOrderId::from("O-001"));
3272                assert_eq!(b.events[1].client_order_id, ClientOrderId::from("O-002"));
3273            }
3274            _ => panic!("Expected OrderAcceptedBatch event"),
3275        }
3276    }
3277
3278    #[tokio::test]
3279    async fn test_execution_event_order_canceled_batch_channel() {
3280        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
3281
3282        let events = vec![
3283            OrderCanceledSpec::builder()
3284                .client_order_id(ClientOrderId::from("O-001"))
3285                .build(),
3286            OrderCanceledSpec::builder()
3287                .client_order_id(ClientOrderId::from("O-002"))
3288                .build(),
3289        ];
3290
3291        let batch = OrderCanceledBatch::new(events);
3292        tx.send(ExecutionEvent::OrderCanceledBatch(batch)).unwrap();
3293
3294        let received = rx.recv().await.unwrap();
3295        match received {
3296            ExecutionEvent::OrderCanceledBatch(b) => {
3297                assert_eq!(b.len(), 2);
3298                assert_eq!(b.events[0].client_order_id, ClientOrderId::from("O-001"));
3299                assert_eq!(b.events[1].client_order_id, ClientOrderId::from("O-002"));
3300            }
3301            _ => panic!("Expected OrderCanceledBatch event"),
3302        }
3303    }
3304
3305    #[cfg(feature = "defi")]
3306    #[rstest]
3307    fn test_handle_data_event_routes_pool_snapshot_response_to_data_engine() {
3308        use nautilus_common::{
3309            messages::defi::PoolSnapshotResponse, msgbus::ShareableMessageHandler,
3310        };
3311        use nautilus_model::defi::{
3312            data::block::BlockPosition,
3313            pool_analysis::snapshot::{PoolAnalytics, PoolSnapshot, PoolState},
3314        };
3315
3316        let received = Rc::new(RefCell::new(Vec::new()));
3317        let sink = received.clone();
3318        msgbus::register_any(
3319            MessagingSwitchboard::data_engine_process(),
3320            ShareableMessageHandler::from_typed(move |response: &PoolSnapshotResponse| {
3321                sink.borrow_mut().push(response.correlation_id);
3322            }),
3323        );
3324
3325        let correlation_id = UUID4::new();
3326
3327        let snapshot = PoolSnapshot::new(
3328            InstrumentId::from("0x11b815efB8f581194ae79006d24E0d814B7697F6.Arbitrum:UniswapV3"),
3329            PoolState::default(),
3330            Vec::new(),
3331            Vec::new(),
3332            PoolAnalytics::default(),
3333            BlockPosition::new(1000, "0x0".to_string(), 0, 0),
3334            UnixNanos::default(),
3335            UnixNanos::default(),
3336        );
3337
3338        AsyncRunner::handle_data_event(DataEvent::PoolSnapshotResponse(PoolSnapshotResponse::new(
3339            correlation_id,
3340            snapshot,
3341        )));
3342
3343        assert_eq!(*received.borrow(), vec![correlation_id]);
3344    }
3345}