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

nautilus_system/
controller.rs
