nautilus_system/
controller.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
16use std::{cell::RefCell, fmt::Debug, rc::Rc};
17
18use nautilus_common::{
19    actor::{
20        DataActor, DataActorCore, DataActorNative, data_actor::DataActorConfig,
21        registry::try_get_actor_unchecked,
22    },
23    component::Component,
24    msgbus::{Endpoint, MStr, TypedHandler, get_message_bus},
25    nautilus_actor,
26};
27use nautilus_model::identifiers::{ActorId, StrategyId};
28use nautilus_trading::{Strategy, StrategyNative};
29
30use crate::{messages::ControllerCommand, trader::Trader};
31
32#[derive(Debug)]
33pub struct Controller {
34    core: DataActorCore,
35    trader: Rc<RefCell<Trader>>,
36}
37
38impl Controller {
39    pub const EXECUTE_ENDPOINT: &str = "Controller.execute";
40
41    #[must_use]
42    pub fn new(trader: Rc<RefCell<Trader>>, config: Option<DataActorConfig>) -> Self {
43        Self {
44            core: DataActorCore::new(config.unwrap_or_default()),
45            trader,
46        }
47    }
48
49    /// Sends a controller command to the registered controller endpoint.
50    ///
51    /// # Errors
52    ///
53    /// Returns an error if the controller execute endpoint is not registered.
54    pub fn send(command: &ControllerCommand) -> anyhow::Result<()> {
55        let endpoint = Self::execute_endpoint();
56        let handler = {
57            let msgbus = get_message_bus();
58            msgbus
59                .borrow_mut()
60                .endpoint_map::<ControllerCommand>()
61                .get(endpoint)
62                .cloned()
63        };
64
65        let Some(handler) = handler else {
66            anyhow::bail!(
67                "Controller execute endpoint '{}' not registered",
68                endpoint.as_str()
69            );
70        };
71
72        handler.handle(command);
73        Ok(())
74    }
75
76    /// Executes a controller command against the underlying trader.
77    ///
78    /// # Errors
79    ///
80    /// Returns an error if the requested lifecycle operation fails.
81    pub fn execute(&mut self, command: ControllerCommand) -> anyhow::Result<()> {
82        match command {
83            ControllerCommand::CreateActor(command) => {
84                Self::unsupported_create_actor_command(&command)
85            }
86            ControllerCommand::StartActor(command) => self.start_actor(&command.actor_id),
87            ControllerCommand::StopActor(command) => self.stop_actor(&command.actor_id),
88            ControllerCommand::RemoveActor(command) => self.remove_actor(&command.actor_id),
89            ControllerCommand::CreateStrategy(command) => {
90                Self::unsupported_create_strategy_command(&command)
91            }
92            ControllerCommand::StartStrategy(command) => self.start_strategy(&command.strategy_id),
93            ControllerCommand::StopStrategy(command) => self.stop_strategy(&command.strategy_id),
94            ControllerCommand::ExitMarket(strategy_id) => self.exit_market(&strategy_id),
95            ControllerCommand::RemoveStrategy(command) => {
96                self.remove_strategy(&command.strategy_id)
97            }
98        }
99    }
100
101    /// Creates a new actor and optionally starts it.
102    ///
103    /// # Errors
104    ///
105    /// Returns an error if actor registration or startup fails.
106    pub fn create_actor<T>(&self, actor: T, start: bool) -> anyhow::Result<ActorId>
107    where
108        T: DataActor + DataActorNative + Component + Debug + 'static,
109    {
110        let actor_id = actor.actor_id();
111        self.trader.borrow_mut().add_actor(actor)?;
112
113        self.start_created_actor(actor_id, start)?;
114
115        Ok(actor_id)
116    }
117
118    /// Creates a new actor from a factory and optionally starts it.
119    ///
120    /// # Errors
121    ///
122    /// Returns an error if the factory, actor registration, or startup fails.
123    pub fn create_actor_from_factory<F, T>(
124        &self,
125        factory: F,
126        start: bool,
127    ) -> anyhow::Result<ActorId>
128    where
129        F: FnOnce() -> anyhow::Result<T>,
130        T: DataActor + DataActorNative + Component + Debug + 'static,
131    {
132        let actor = factory()?;
133        self.create_actor(actor, start)
134    }
135
136    /// Creates a new strategy and optionally starts it.
137    ///
138    /// # Errors
139    ///
140    /// Returns an error if strategy registration or startup fails.
141    pub fn create_strategy<T>(&self, mut strategy: T, start: bool) -> anyhow::Result<StrategyId>
142    where
143        T: Strategy + StrategyNative + DataActorNative + Component + Debug + 'static,
144    {
145        let strategy_id = self
146            .trader
147            .borrow()
148            .prepare_strategy_for_registration(&mut strategy)?;
149        self.trader.borrow_mut().add_strategy(strategy)?;
150
151        self.start_created_strategy(strategy_id, start)?;
152
153        Ok(strategy_id)
154    }
155
156    /// Creates a new strategy from a factory and optionally starts it.
157    ///
158    /// # Errors
159    ///
160    /// Returns an error if the factory, strategy registration, or startup fails.
161    pub fn create_strategy_from_factory<F, T>(
162        &self,
163        factory: F,
164        start: bool,
165    ) -> anyhow::Result<StrategyId>
166    where
167        F: FnOnce() -> anyhow::Result<T>,
168        T: Strategy + StrategyNative + DataActorNative + Component + Debug + 'static,
169    {
170        let strategy = factory()?;
171        self.create_strategy(strategy, start)
172    }
173
174    /// Starts the registered actor with the given identifier.
175    ///
176    /// # Errors
177    ///
178    /// Returns an error if the actor is not registered or cannot be started.
179    pub fn start_actor(&self, actor_id: &ActorId) -> anyhow::Result<()> {
180        self.trader.borrow().start_actor(actor_id)
181    }
182
183    /// Stops the registered actor with the given identifier.
184    ///
185    /// # Errors
186    ///
187    /// Returns an error if the actor is not registered or cannot be stopped.
188    pub fn stop_actor(&self, actor_id: &ActorId) -> anyhow::Result<()> {
189        self.trader.borrow().stop_actor(actor_id)
190    }
191
192    /// Removes the registered actor with the given identifier.
193    ///
194    /// # Errors
195    ///
196    /// Returns an error if the actor cannot be removed.
197    pub fn remove_actor(&self, actor_id: &ActorId) -> anyhow::Result<()> {
198        if actor_id.inner() == self.core.actor_id().inner() {
199            return Ok(());
200        }
201
202        self.trader.borrow_mut().remove_actor(actor_id)
203    }
204
205    /// Starts the registered strategy with the given identifier.
206    ///
207    /// # Errors
208    ///
209    /// Returns an error if the strategy is not registered or cannot be started.
210    pub fn start_strategy(&self, strategy_id: &StrategyId) -> anyhow::Result<()> {
211        self.trader.borrow().start_strategy(strategy_id)
212    }
213
214    /// Stops the registered strategy with the given identifier.
215    ///
216    /// # Errors
217    ///
218    /// Returns an error if the strategy is not registered or cannot be stopped.
219    pub fn stop_strategy(&self, strategy_id: &StrategyId) -> anyhow::Result<()> {
220        self.trader.borrow_mut().stop_strategy(strategy_id)
221    }
222
223    /// Sends an exit-market command to the registered strategy.
224    ///
225    /// # Errors
226    ///
227    /// Returns an error if the strategy is not registered or its control endpoint is missing.
228    pub fn exit_market(&self, strategy_id: &StrategyId) -> anyhow::Result<()> {
229        Trader::market_exit_strategy(&self.trader, strategy_id)
230    }
231
232    /// Removes the registered strategy with the given identifier.
233    ///
234    /// # Errors
235    ///
236    /// Returns an error if the strategy cannot be removed.
237    pub fn remove_strategy(&self, strategy_id: &StrategyId) -> anyhow::Result<()> {
238        self.trader.borrow_mut().remove_strategy(strategy_id)
239    }
240
241    fn start_created_actor(&self, actor_id: ActorId, start: bool) -> anyhow::Result<()> {
242        if !start {
243            return Ok(());
244        }
245
246        if let Err(start_err) = self.start_actor(&actor_id) {
247            return Err(self.rollback_actor_start_failure(actor_id, start_err));
248        }
249
250        Ok(())
251    }
252
253    fn start_created_strategy(&self, strategy_id: StrategyId, start: bool) -> anyhow::Result<()> {
254        if !start {
255            return Ok(());
256        }
257
258        if let Err(start_err) = self.start_strategy(&strategy_id) {
259            return Err(self.rollback_strategy_start_failure(strategy_id, start_err));
260        }
261
262        Ok(())
263    }
264
265    fn rollback_actor_start_failure(
266        &self,
267        actor_id: ActorId,
268        start_err: anyhow::Error,
269    ) -> anyhow::Error {
270        match self.remove_actor(&actor_id) {
271            Ok(()) => start_err,
272            Err(rollback_err) => anyhow::anyhow!(
273                "Failed to start actor {actor_id}: {start_err}; rollback failed: {rollback_err}"
274            ),
275        }
276    }
277
278    fn rollback_strategy_start_failure(
279        &self,
280        strategy_id: StrategyId,
281        start_err: anyhow::Error,
282    ) -> anyhow::Error {
283        match self.remove_strategy(&strategy_id) {
284            Ok(()) => start_err,
285            Err(rollback_err) => anyhow::anyhow!(
286                "Failed to start strategy {strategy_id}: {start_err}; rollback failed: {rollback_err}"
287            ),
288        }
289    }
290
291    fn register_execute_endpoint(&self) {
292        let controller_id = self.core.actor_id().inner();
293        let handler = TypedHandler::from(move |command: &ControllerCommand| {
294            if let Some(mut controller) = try_get_actor_unchecked::<Self>(&controller_id) {
295                if let Err(e) = controller.execute(command.clone()) {
296                    log::error!("Controller command failed for {controller_id}: {e}");
297                }
298            } else {
299                log::error!("Controller {controller_id} not found for command handling");
300            }
301        });
302
303        get_message_bus()
304            .borrow_mut()
305            .endpoint_map::<ControllerCommand>()
306            .register(Self::execute_endpoint(), handler);
307    }
308
309    fn deregister_execute_endpoint() {
310        get_message_bus()
311            .borrow_mut()
312            .endpoint_map::<ControllerCommand>()
313            .deregister(Self::execute_endpoint());
314    }
315
316    fn execute_endpoint() -> MStr<Endpoint> {
317        Self::EXECUTE_ENDPOINT.into()
318    }
319
320    fn unsupported_create_actor_command(
321        command: &crate::messages::CreateActor,
322    ) -> anyhow::Result<()> {
323        anyhow::bail!(
324            "CreateActor command for importable actor '{}' is not supported by the Rust controller",
325            command.actor_config.actor_path
326        );
327    }
328
329    fn unsupported_create_strategy_command(
330        command: &crate::messages::CreateStrategy,
331    ) -> anyhow::Result<()> {
332        anyhow::bail!(
333            "CreateStrategy command for importable strategy '{}' is not supported by the Rust controller",
334            command.strategy_config.strategy_path
335        );
336    }
337}
338
339impl DataActor for Controller {
340    fn on_start(&mut self) -> anyhow::Result<()> {
341        self.register_execute_endpoint();
342        Ok(())
343    }
344
345    fn on_stop(&mut self) -> anyhow::Result<()> {
346        Self::deregister_execute_endpoint();
347        Ok(())
348    }
349
350    fn on_resume(&mut self) -> anyhow::Result<()> {
351        self.register_execute_endpoint();
352        Ok(())
353    }
354
355    fn on_dispose(&mut self) -> anyhow::Result<()> {
356        Self::deregister_execute_endpoint();
357        Ok(())
358    }
359}
360
361nautilus_actor!(Controller);
362
363#[cfg(test)]
364mod tests {
365    use std::collections::HashMap;
366
367    use nautilus_common::{
368        actor::data_actor::ImportableActorConfig,
369        cache::Cache,
370        clock::TestClock,
371        enums::{ComponentState, Environment},
372        msgbus::{MessageBus, set_message_bus},
373    };
374    use nautilus_core::{UUID4, UnixNanos};
375    use nautilus_model::{identifiers::TraderId, stubs::TestDefault};
376    use nautilus_portfolio::portfolio::Portfolio;
377    use nautilus_trading::{
378        ImportableStrategyConfig, nautilus_strategy,
379        strategy::{StrategyConfig, StrategyCore},
380    };
381    use rstest::rstest;
382
383    use super::*;
384    use crate::{
385        clock_factory::ClockFactory,
386        messages::{
387            CreateActor, CreateStrategy, RemoveActor, RemoveStrategy, StartActor, StartStrategy,
388            StopActor, StopStrategy,
389        },
390    };
391
392    fn start_actor_command(actor_id: ActorId) -> ControllerCommand {
393        StartActor::new(actor_id, UUID4::new(), UnixNanos::default()).into()
394    }
395
396    fn stop_actor_command(actor_id: ActorId) -> ControllerCommand {
397        StopActor::new(actor_id, UUID4::new(), UnixNanos::default()).into()
398    }
399
400    fn remove_actor_command(actor_id: ActorId) -> ControllerCommand {
401        RemoveActor::new(actor_id, UUID4::new(), UnixNanos::default()).into()
402    }
403
404    fn start_strategy_command(strategy_id: StrategyId) -> ControllerCommand {
405        StartStrategy::new(strategy_id, UUID4::new(), UnixNanos::default()).into()
406    }
407
408    fn stop_strategy_command(strategy_id: StrategyId) -> ControllerCommand {
409        StopStrategy::new(strategy_id, UUID4::new(), UnixNanos::default()).into()
410    }
411
412    fn remove_strategy_command(strategy_id: StrategyId) -> ControllerCommand {
413        RemoveStrategy::new(strategy_id, UUID4::new(), UnixNanos::default()).into()
414    }
415
416    #[derive(Debug)]
417    struct TestDataActor {
418        core: DataActorCore,
419    }
420
421    impl TestDataActor {
422        fn new(config: DataActorConfig) -> Self {
423            Self {
424                core: DataActorCore::new(config),
425            }
426        }
427    }
428
429    impl DataActor for TestDataActor {}
430
431    nautilus_actor!(TestDataActor);
432
433    #[derive(Debug)]
434    struct TestStrategy {
435        core: StrategyCore,
436    }
437
438    impl TestStrategy {
439        fn new(config: StrategyConfig) -> Self {
440            Self {
441                core: StrategyCore::new(config),
442            }
443        }
444    }
445
446    impl DataActor for TestStrategy {}
447
448    nautilus_strategy!(TestStrategy);
449
450    #[derive(Debug)]
451    struct FailingStartActor {
452        core: DataActorCore,
453    }
454
455    impl FailingStartActor {
456        fn new(config: DataActorConfig) -> Self {
457            Self {
458                core: DataActorCore::new(config),
459            }
460        }
461    }
462
463    impl DataActor for FailingStartActor {
464        fn on_start(&mut self) -> anyhow::Result<()> {
465            anyhow::bail!("Simulated actor start failure")
466        }
467    }
468
469    nautilus_actor!(FailingStartActor);
470
471    #[derive(Debug)]
472    struct FailingStartStrategy {
473        core: StrategyCore,
474    }
475
476    impl FailingStartStrategy {
477        fn new(config: StrategyConfig) -> Self {
478            Self {
479                core: StrategyCore::new(config),
480            }
481        }
482    }
483
484    impl DataActor for FailingStartStrategy {
485        fn on_start(&mut self) -> anyhow::Result<()> {
486            anyhow::bail!("Simulated strategy start failure")
487        }
488    }
489
490    nautilus_strategy!(FailingStartStrategy);
491
492    #[derive(Debug)]
493    struct ReentrantExitStrategy {
494        core: StrategyCore,
495        actor_to_stop: ActorId,
496    }
497
498    impl ReentrantExitStrategy {
499        fn new(config: StrategyConfig, actor_to_stop: ActorId) -> Self {
500            Self {
501                core: StrategyCore::new(config),
502                actor_to_stop,
503            }
504        }
505    }
506
507    impl DataActor for ReentrantExitStrategy {}
508
509    nautilus_strategy!(ReentrantExitStrategy, {
510        fn on_market_exit(&mut self) {
511            Controller::send(&stop_actor_command(self.actor_to_stop)).unwrap();
512        }
513    });
514
515    fn create_running_controller() -> (Rc<RefCell<Trader>>, ActorId) {
516        let trader_id = TraderId::test_default();
517        let instance_id = UUID4::new();
518        let clock_factory = ClockFactory::test_default();
519        let clock = clock_factory.clock();
520        let mut clock_ref = clock.borrow_mut();
521        let test_clock = clock_ref
522            .as_any_mut()
523            .downcast_mut::<TestClock>()
524            .expect("test default clock must be TestClock");
525        test_clock.set_time(1_000_000_000u64.into());
526        drop(clock_ref);
527
528        let msgbus = Rc::new(RefCell::new(MessageBus::new(
529            trader_id,
530            instance_id,
531            Some("test".to_string()),
532            None,
533        )));
534        set_message_bus(msgbus);
535
536        let cache = Rc::new(RefCell::new(Cache::new(None, None)));
537        let portfolio = Rc::new(RefCell::new(Portfolio::new(
538            clock.clone(),
539            cache.clone(),
540            None,
541        )));
542
543        let trader = Rc::new(RefCell::new(Trader::new(
544            trader_id,
545            instance_id,
546            Environment::Backtest,
547            clock_factory,
548            cache,
549            portfolio,
550        )));
551        trader.borrow_mut().initialize().unwrap();
552
553        let controller = Controller::new(
554            trader.clone(),
555            Some(DataActorConfig {
556                actor_id: Some(ActorId::from("Controller-001")),
557                ..Default::default()
558            }),
559        );
560        let controller_id = controller.core.actor_id();
561
562        trader.borrow_mut().add_actor(controller).unwrap();
563        trader.borrow_mut().start().unwrap();
564
565        (trader, controller_id)
566    }
567
568    #[rstest]
569    fn test_controller_rejects_importable_create_commands() {
570        let (trader, controller_id) = create_running_controller();
571        let controller_actor_id = controller_id.inner();
572
573        let mut controller = try_get_actor_unchecked::<Controller>(&controller_actor_id).unwrap();
574        let actor_config = ImportableActorConfig {
575            actor_path: "tests.actors:Actor".to_string(),
576            config_path: "tests.actors:ActorConfig".to_string(),
577            config: HashMap::new(),
578        };
579        let strategy_config = ImportableStrategyConfig {
580            strategy_path: "tests.strategies:Strategy".to_string(),
581            config_path: "tests.strategies:StrategyConfig".to_string(),
582            config: HashMap::new(),
583        };
584
585        let actor_result = controller.execute(
586            CreateActor::new(actor_config, true, UUID4::new(), UnixNanos::default()).into(),
587        );
588        let strategy_result = controller.execute(
589            CreateStrategy::new(strategy_config, true, UUID4::new(), UnixNanos::default()).into(),
590        );
591
592        assert_eq!(
593            actor_result.unwrap_err().to_string(),
594            "CreateActor command for importable actor 'tests.actors:Actor' is not supported by the Rust controller"
595        );
596        assert_eq!(
597            strategy_result.unwrap_err().to_string(),
598            "CreateStrategy command for importable strategy 'tests.strategies:Strategy' is not supported by the Rust controller"
599        );
600
601        drop(controller);
602        trader.borrow_mut().stop().unwrap();
603        trader.borrow_mut().dispose_components().unwrap();
604    }
605
606    #[rstest]
607    fn test_controller_manages_actor_lifecycle_by_message() {
608        let (trader, controller_id) = create_running_controller();
609        let controller_actor_id = controller_id.inner();
610
611        let actor_id = {
612            let controller = try_get_actor_unchecked::<Controller>(&controller_actor_id).unwrap();
613            controller
614                .create_actor(
615                    TestDataActor::new(DataActorConfig {
616                        actor_id: Some(ActorId::from("TestActor-001")),
617                        ..Default::default()
618                    }),
619                    false,
620                )
621                .unwrap()
622        };
623
624        assert!(trader.borrow().actor_ids().contains(&actor_id));
625
626        Controller::send(&start_actor_command(actor_id)).unwrap();
627        let actor_registry_id = actor_id.inner();
628        assert_eq!(
629            try_get_actor_unchecked::<TestDataActor>(&actor_registry_id)
630                .unwrap()
631                .state(),
632            ComponentState::Running
633        );
634
635        Controller::send(&stop_actor_command(actor_id)).unwrap();
636        assert_eq!(
637            try_get_actor_unchecked::<TestDataActor>(&actor_registry_id)
638                .unwrap()
639                .state(),
640            ComponentState::Stopped
641        );
642
643        Controller::send(&remove_actor_command(actor_id)).unwrap();
644        assert!(!trader.borrow().actor_ids().contains(&actor_id));
645
646        trader.borrow_mut().stop().unwrap();
647        trader.borrow_mut().dispose_components().unwrap();
648    }
649
650    #[rstest]
651    fn test_controller_manages_strategy_lifecycle_and_exit_market() {
652        let (trader, controller_id) = create_running_controller();
653        let controller_actor_id = controller_id.inner();
654
655        let strategy_id = {
656            let controller = try_get_actor_unchecked::<Controller>(&controller_actor_id).unwrap();
657            controller
658                .create_strategy(
659                    TestStrategy::new(StrategyConfig {
660                        strategy_id: Some(StrategyId::from("TestStrategy-001")),
661                        order_id_tag: Some("001".to_string()),
662                        ..Default::default()
663                    }),
664                    false,
665                )
666                .unwrap()
667        };
668
669        assert!(trader.borrow().strategy_ids().contains(&strategy_id));
670
671        Controller::send(&start_strategy_command(strategy_id)).unwrap();
672        let strategy_registry_id = strategy_id.inner();
673        assert_eq!(
674            try_get_actor_unchecked::<TestStrategy>(&strategy_registry_id)
675                .unwrap()
676                .state(),
677            ComponentState::Running
678        );
679
680        Controller::send(&ControllerCommand::ExitMarket(strategy_id)).unwrap();
681        assert!(
682            try_get_actor_unchecked::<TestStrategy>(&strategy_registry_id)
683                .unwrap()
684                .is_exiting()
685        );
686
687        Controller::send(&stop_strategy_command(strategy_id)).unwrap();
688        let strategy = try_get_actor_unchecked::<TestStrategy>(&strategy_registry_id).unwrap();
689        assert_eq!(strategy.state(), ComponentState::Stopped);
690        assert!(!strategy.is_exiting());
691        drop(strategy);
692
693        Controller::send(&remove_strategy_command(strategy_id)).unwrap();
694        assert!(!trader.borrow().strategy_ids().contains(&strategy_id));
695
696        trader.borrow_mut().stop().unwrap();
697        trader.borrow_mut().dispose_components().unwrap();
698    }
699
700    #[rstest]
701    fn test_controller_create_actor_rolls_back_on_start_failure() {
702        let (trader, controller_id) = create_running_controller();
703        let controller_actor_id = controller_id.inner();
704        let actor_id = ActorId::from("FailingActor-001");
705
706        let result = {
707            let controller = try_get_actor_unchecked::<Controller>(&controller_actor_id).unwrap();
708            controller.create_actor(
709                FailingStartActor::new(DataActorConfig {
710                    actor_id: Some(actor_id),
711                    ..Default::default()
712                }),
713                true,
714            )
715        };
716
717        assert!(result.is_err());
718        assert!(
719            result
720                .unwrap_err()
721                .to_string()
722                .contains("Simulated actor start failure")
723        );
724        assert!(!trader.borrow().actor_ids().contains(&actor_id));
725        if let Some(actor) = try_get_actor_unchecked::<FailingStartActor>(&actor_id.inner()) {
726            assert_eq!(actor.state(), ComponentState::Disposed);
727        }
728
729        trader.borrow_mut().stop().unwrap();
730        trader.borrow_mut().dispose_components().unwrap();
731    }
732
733    #[rstest]
734    fn test_controller_create_strategy_rolls_back_on_start_failure() {
735        let (trader, controller_id) = create_running_controller();
736        let controller_actor_id = controller_id.inner();
737        let strategy_id = StrategyId::from("FailingStrategy-001");
738
739        let result = {
740            let controller = try_get_actor_unchecked::<Controller>(&controller_actor_id).unwrap();
741            controller.create_strategy(
742                FailingStartStrategy::new(StrategyConfig {
743                    strategy_id: Some(strategy_id),
744                    order_id_tag: Some("001".to_string()),
745                    ..Default::default()
746                }),
747                true,
748            )
749        };
750
751        assert!(result.is_err());
752        assert!(
753            result
754                .unwrap_err()
755                .to_string()
756                .contains("Simulated strategy start failure")
757        );
758        assert!(!trader.borrow().strategy_ids().contains(&strategy_id));
759
760        if let Some(strategy) =
761            try_get_actor_unchecked::<FailingStartStrategy>(&strategy_id.inner())
762        {
763            assert_eq!(strategy.state(), ComponentState::Disposed);
764        }
765
766        trader.borrow_mut().stop().unwrap();
767        trader.borrow_mut().dispose_components().unwrap();
768    }
769
770    #[rstest]
771    fn test_controller_exit_market_allows_reentrant_controller_commands() {
772        let (trader, controller_id) = create_running_controller();
773        let controller_actor_id = controller_id.inner();
774
775        let helper_actor_id = {
776            let controller = try_get_actor_unchecked::<Controller>(&controller_actor_id).unwrap();
777            controller
778                .create_actor(
779                    TestDataActor::new(DataActorConfig {
780                        actor_id: Some(ActorId::from("HelperActor-001")),
781                        ..Default::default()
782                    }),
783                    true,
784                )
785                .unwrap()
786        };
787
788        let strategy_id = {
789            let controller = try_get_actor_unchecked::<Controller>(&controller_actor_id).unwrap();
790            controller
791                .create_strategy(
792                    ReentrantExitStrategy::new(
793                        StrategyConfig {
794                            strategy_id: Some(StrategyId::from("ReentrantStrategy-001")),
795                            order_id_tag: Some("001".to_string()),
796                            ..Default::default()
797                        },
798                        helper_actor_id,
799                    ),
800                    false,
801                )
802                .unwrap()
803        };
804
805        Controller::send(&start_strategy_command(strategy_id)).unwrap();
806        Controller::send(&ControllerCommand::ExitMarket(strategy_id)).unwrap();
807
808        let helper_actor =
809            try_get_actor_unchecked::<TestDataActor>(&helper_actor_id.inner()).unwrap();
810        assert_eq!(helper_actor.state(), ComponentState::Stopped);
811        drop(helper_actor);
812        assert!(
813            try_get_actor_unchecked::<ReentrantExitStrategy>(&strategy_id.inner())
814                .unwrap()
815                .is_exiting()
816        );
817
818        Controller::send(&stop_strategy_command(strategy_id)).unwrap();
819        Controller::send(&remove_strategy_command(strategy_id)).unwrap();
820        Controller::send(&remove_actor_command(helper_actor_id)).unwrap();
821        trader.borrow_mut().stop().unwrap();
822        trader.borrow_mut().dispose_components().unwrap();
823    }
824
825    #[rstest]
826    fn test_controller_send_fails_after_controller_stop() {
827        let (trader, _) = create_running_controller();
828
829        trader.borrow_mut().stop().unwrap();
830
831        let result = Controller::send(&stop_actor_command(ActorId::from("AnyActor-001")));
832        assert!(result.is_err());
833        assert_eq!(
834            result.unwrap_err().to_string(),
835            "Controller execute endpoint 'Controller.execute' not registered"
836        );
837
838        trader.borrow_mut().dispose_components().unwrap();
839    }
840}
nautilus_system/controller.rs

nautilus_system/
controller.rs
