Wire OrderEmulator lifecycle paths

- Start emulator replay through trader startup
- Reset and dispose emulator state through kernel lifecycle paths
- Reset emulator state in the direct backtest reset path
- Skip reentrant emulator order events with clearer logging

Author: cjdsellers

crates/backtest/src/engine.rs

@@ -891,6 +891,8 @@ impl BacktestEngine {
         self.kernel.risk_engine.borrow_mut().stop();
         self.kernel.risk_engine.borrow_mut().reset();
 
+        self.kernel.order_emulator.reset();
+
         // Reset trader
         if let Err(e) = self.kernel.trader.borrow_mut().reset() {
             log::error!("Error resetting trader: {e:?}");
@@ -1840,19 +1842,34 @@ fn log_portfolio_performance(analyzer: &PortfolioAnalyzer) {
 
 #[cfg(test)]
 mod tests {
-    use nautilus_common::enums::Environment;
+    use nautilus_common::{
+        enums::Environment,
+        messages::{
+            data::{DataCommand, UnsubscribeCommand},
+            execution::SubmitOrder,
+        },
+        msgbus::{
+            self, MessagingSwitchboard,
+            stubs::{TypedIntoMessageSavingHandler, get_typed_into_message_saving_handler},
+        },
+    };
     use nautilus_execution::engine::SnapshotAnchorer;
     use nautilus_model::{
         data::{Data, InstrumentStatus},
-        enums::{AccountType, BookType, MarketStatus, MarketStatusAction, OmsType},
+        enums::{
+            AccountType, BookType, MarketStatus, MarketStatusAction, OmsType, OrderSide, OrderType,
+            TriggerType,
+        },
         identifiers::Venue,
         instruments::{
             CryptoPerpetual, Instrument, InstrumentAny, stubs::crypto_perpetual_ethusdt,
         },
-        types::Money,
+        orders::{Order, OrderAny, OrderTestBuilder},
+        types::{Money, Price, Quantity},
     };
     use nautilus_system::{KernelEventStore, RegisteredComponents};
     use rstest::*;
+    use ustr::Ustr;
 
     use super::*;
 
@@ -1942,6 +1959,43 @@ mod tests {
         engine
     }
 
+    fn create_stop_market_order(instrument: &CryptoPerpetual) -> OrderAny {
+        OrderTestBuilder::new(OrderType::StopMarket)
+            .instrument_id(instrument.id())
+            .side(OrderSide::Buy)
+            .trigger_price(Price::from("5100.00"))
+            .quantity(Quantity::from(1))
+            .emulation_trigger(TriggerType::BidAsk)
+            .build()
+    }
+
+    fn create_submit_order_command(order: &OrderAny) -> SubmitOrder {
+        SubmitOrder::new(
+            order.trader_id(),
+            None,
+            order.strategy_id(),
+            order.instrument_id(),
+            order.client_order_id(),
+            order.init_event().clone(),
+            order.exec_algorithm_id(),
+            None,
+            None,
+            UUID4::new(),
+            0.into(),
+            None, // correlation_id
+        )
+    }
+
+    fn register_data_command_handler(id: &str) -> TypedIntoMessageSavingHandler<DataCommand> {
+        let (handler, saving_handler) =
+            get_typed_into_message_saving_handler::<DataCommand>(Some(Ustr::from(id)));
+        msgbus::register_data_command_endpoint(
+            MessagingSwitchboard::data_engine_queue_execute(),
+            handler,
+        );
+        saving_handler
+    }
+
     #[rstest]
     fn test_run_impl_event_store_replay_skips_trader_start() {
         let mut engine = create_engine_with_replay_store(false);
@@ -2028,6 +2082,43 @@ mod tests {
         );
     }
 
+    #[rstest]
+    fn test_reset_resets_order_emulator_state(crypto_perpetual_ethusdt: CryptoPerpetual) {
+        let mut engine = create_engine();
+        let data_commands =
+            register_data_command_handler("DataEngine.queue_execute.backtest_reset");
+        let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt.clone());
+        let instrument_id = instrument.id();
+        engine.add_instrument(&instrument).unwrap();
+        let order = create_stop_market_order(&crypto_perpetual_ethusdt);
+        let command = create_submit_order_command(&order);
+        engine
+            .kernel
+            .cache
+            .borrow_mut()
+            .add_order(order, None, None, false)
+            .unwrap();
+        let order_emulator = engine.kernel.order_emulator.emulator();
+        let mut order_emulator = order_emulator.borrow_mut();
+        order_emulator.cache_submit_order_command(command.clone());
+        order_emulator.handle_submit_order(command);
+        drop(order_emulator);
+        data_commands.clear();
+
+        engine.reset();
+
+        let commands = data_commands.get_messages();
+        let emulator = engine.kernel.order_emulator.get_emulator();
+        assert!(emulator.subscribed_quotes().is_empty());
+        assert!(emulator.subscribed_trades().is_empty());
+        assert!(emulator.get_matching_core(&instrument_id).is_none());
+        assert!(commands.iter().any(|command| matches!(
+            command,
+            DataCommand::Unsubscribe(UnsubscribeCommand::Quotes(command))
+                if command.instrument_id == instrument_id
+        )));
+    }
+
     #[rstest]
     fn test_route_data_to_exchange_instrument_status(crypto_perpetual_ethusdt: CryptoPerpetual) {
         let mut engine = create_engine();

crates/execution/src/order_emulator/adapter.rs

@@ -49,4 +49,20 @@ impl OrderEmulatorAdapter {
     pub fn emulator(&self) -> Rc<RefCell<OrderEmulator>> {
         self.emulator.clone()
     }
+
+    pub fn start(&self) {
+        self.emulator.borrow_mut().start();
+    }
+
+    pub fn stop(&self) {
+        self.emulator.borrow().stop();
+    }
+
+    pub fn reset(&self) {
+        self.emulator.borrow_mut().reset();
+    }
+
+    pub fn dispose(&self) {
+        self.emulator.borrow_mut().dispose();
+    }
 }

crates/execution/src/order_emulator/emulator.rs

@@ -339,6 +339,7 @@ impl OrderEmulator {
                 None, // correlation_id
             );
 
+            self.manager.cache_submit_order_command(command.clone());
             self.handle_submit_order(command);
         }
 
@@ -524,8 +525,8 @@ impl OrderEmulator {
             self.dispatch_manager_actions(actions);
             return;
         }
-
-        self.check_monitoring(command.strategy_id, command.position_id);
+        let strategy_id = command.strategy_id;
+        let position_id = command.position_id;
 
         // Get or create matching core
         let trigger_instrument_id = order
@@ -672,6 +673,8 @@ impl OrderEmulator {
             );
         }
 
+        self.check_monitoring(strategy_id, position_id);
+
         // Since we are cloning the matching core, we need to insert it back into the original hashmap
         self.matching_cores
             .insert(trigger_instrument_id, matching_core);
@@ -1575,11 +1578,11 @@ mod tests {
     use nautilus_model::{
         data::{QuoteTick, TradeTick},
         enums::{AggressorSide, OrderSide, OrderType, TriggerType},
-        identifiers::{ClientOrderId, StrategyId, TradeId, TraderId},
+        identifiers::{ClientOrderId, OrderListId, StrategyId, TradeId, TraderId},
         instruments::{
             CryptoPerpetual, Instrument, InstrumentAny, stubs::crypto_perpetual_ethusdt,
         },
-        orders::OrderTestBuilder,
+        orders::{OrderList, OrderTestBuilder},
         types::{Price, Quantity},
     };
     use rstest::{fixture, rstest};
@@ -1631,6 +1634,22 @@ mod tests {
             .build()
     }
 
+    fn create_list_stop_market_order(
+        instrument: &CryptoPerpetual,
+        client_order_id: &str,
+        order_list_id: OrderListId,
+    ) -> OrderAny {
+        OrderTestBuilder::new(OrderType::StopMarket)
+            .instrument_id(instrument.id())
+            .client_order_id(ClientOrderId::from(client_order_id))
+            .order_list_id(order_list_id)
+            .side(OrderSide::Buy)
+            .trigger_price(Price::from("5100.00"))
+            .quantity(Quantity::from(1))
+            .emulation_trigger(TriggerType::BidAsk)
+            .build()
+    }
+
     fn create_submit_order(instrument: &CryptoPerpetual, order: &OrderAny) -> SubmitOrder {
         SubmitOrder::new(
             TraderId::from("TRADER-001"),
@@ -2163,6 +2182,70 @@ mod tests {
         assert_eq!(core.last, None);
     }
 
+    #[rstest]
+    fn test_submit_order_list_handles_reentrant_order_events(instrument: CryptoPerpetual) {
+        let (_clock, cache, emulator) = create_emulator();
+        let data_commands = register_data_command_handler("DataEngine.queue_execute.list");
+        add_instrument_to_cache(&cache, &instrument);
+        let order_list_id = OrderListId::from("OL-EMULATOR-001");
+        let first_order = create_list_stop_market_order(&instrument, "O-LIST-001", order_list_id);
+        let second_order = create_list_stop_market_order(&instrument, "O-LIST-002", order_list_id);
+        let orders = vec![first_order.clone(), second_order.clone()];
+        let order_list = OrderList::from_orders(&orders, 0.into());
+        let order_inits = orders
+            .iter()
+            .map(|order| order.init_event().clone())
+            .collect();
+        cache
+            .borrow_mut()
+            .add_order(first_order.clone(), None, None, false)
+            .unwrap();
+        cache
+            .borrow_mut()
+            .add_order(second_order.clone(), None, None, false)
+            .unwrap();
+        let command = SubmitOrderList::new(
+            TraderId::from("TRADER-001"),
+            None,
+            StrategyId::from("STRATEGY-001"),
+            order_list,
+            order_inits,
+            None,
+            None,
+            None,
+            UUID4::new(),
+            0.into(),
+            None,
+        );
+
+        emulator
+            .borrow_mut()
+            .execute(TradingCommand::SubmitOrderList(command));
+
+        let commands = data_commands.get_messages();
+        let cache = cache.borrow();
+        let first_status = cache
+            .order(&first_order.client_order_id())
+            .unwrap()
+            .status();
+        let second_status = cache
+            .order(&second_order.client_order_id())
+            .unwrap()
+            .status();
+        drop(cache);
+        let emulator = emulator.borrow();
+
+        assert_eq!(first_status, OrderStatus::Emulated);
+        assert_eq!(second_status, OrderStatus::Emulated);
+        assert!(emulator.get_matching_core(&instrument.id()).is_some());
+        assert_eq!(emulator.subscribed_quotes(), vec![instrument.id()]);
+        assert!(commands.iter().any(|command| matches!(
+            command,
+            DataCommand::Subscribe(SubscribeCommand::Quotes(command))
+                if command.instrument_id == instrument.id()
+        )));
+    }
+
     #[rstest]
     fn test_submit_order_caches_command(instrument: CryptoPerpetual) {
         let (_clock, cache, emulator) = create_emulator();

crates/execution/src/order_emulator/handlers.rs

@@ -73,7 +73,18 @@ impl Handler<OrderEventAny> for OrderEmulatorOnEventHandler {
 
     fn handle(&self, event: &OrderEventAny) {
         if let Some(emulator) = self.emulator.upgrade() {
-            emulator.borrow_mut().on_event(event);
+            match emulator.try_borrow_mut() {
+                Ok(mut emulator) => emulator.on_event(event),
+                Err(_) => {
+                    log::debug!(
+                        concat!(
+                            "Skipping reentrant order event while OrderEmulator is already handling ",
+                            "a command or event; expected for self-published emulator events: {}"
+                        ),
+                        event
+                    );
+                }
+            }
         }
     }
 }