1#![allow(unsafe_code)]
23
24use std::{
25 cell::{RefCell, UnsafeCell},
26 fmt::Debug,
27 rc::Rc,
28};
29
30use ahash::{AHashMap, AHashSet};
31use nautilus_model::identifiers::{ComponentId, TraderId};
32use ustr::Ustr;
33
34use crate::{
35 actor::{Actor, registry::with_actor_registry},
36 cache::Cache,
37 clock::Clock,
38 enums::{ComponentState, ComponentTrigger},
39};
40
41pub trait Component {
43 fn component_id(&self) -> ComponentId;
45
46 fn state(&self) -> ComponentState;
48
49 fn transition_state(&mut self, trigger: ComponentTrigger) -> anyhow::Result<()>;
55
56 fn is_ready(&self) -> bool {
58 self.state() == ComponentState::Ready
59 }
60
61 fn not_running(&self) -> bool {
63 !self.is_running()
64 }
65
66 fn is_running(&self) -> bool {
68 self.state() == ComponentState::Running
69 }
70
71 fn is_stopped(&self) -> bool {
73 self.state() == ComponentState::Stopped
74 }
75
76 fn is_degraded(&self) -> bool {
78 self.state() == ComponentState::Degraded
79 }
80
81 fn is_faulted(&self) -> bool {
83 self.state() == ComponentState::Faulted
84 }
85
86 fn is_disposed(&self) -> bool {
88 self.state() == ComponentState::Disposed
89 }
90
91 fn register(
97 &mut self,
98 trader_id: TraderId,
99 clock: Rc<RefCell<dyn Clock>>,
100 cache: Rc<RefCell<Cache>>,
101 ) -> anyhow::Result<()>;
102
103 fn initialize(&mut self) -> anyhow::Result<()> {
109 self.transition_state(ComponentTrigger::Initialize)
110 }
111
112 fn start(&mut self) -> anyhow::Result<()> {
118 self.transition_state(ComponentTrigger::Start)?; if let Err(e) = self.on_start() {
121 log_error(self.component_id(), &e);
122 return Err(e); }
124
125 self.transition_state(ComponentTrigger::StartCompleted)?;
126
127 Ok(())
128 }
129
130 fn stop(&mut self) -> anyhow::Result<()> {
136 self.transition_state(ComponentTrigger::Stop)?; if let Err(e) = self.on_stop() {
139 log_error(self.component_id(), &e);
140 return Err(e); }
142
143 self.transition_state(ComponentTrigger::StopCompleted)?;
144
145 Ok(())
146 }
147
148 fn resume(&mut self) -> anyhow::Result<()> {
154 self.transition_state(ComponentTrigger::Resume)?; if let Err(e) = self.on_resume() {
157 log_error(self.component_id(), &e);
158 return Err(e); }
160
161 self.transition_state(ComponentTrigger::ResumeCompleted)?;
162
163 Ok(())
164 }
165
166 fn degrade(&mut self) -> anyhow::Result<()> {
172 self.transition_state(ComponentTrigger::Degrade)?; if let Err(e) = self.on_degrade() {
175 log_error(self.component_id(), &e);
176 return Err(e); }
178
179 self.transition_state(ComponentTrigger::DegradeCompleted)?;
180
181 Ok(())
182 }
183
184 fn fault(&mut self) -> anyhow::Result<()> {
196 self.transition_state(ComponentTrigger::Fault)?; let result = self.on_fault();
199 self.release_subscriptions();
200
201 if let Err(e) = result {
202 log_error(self.component_id(), &e);
203 return Err(e); }
205
206 self.transition_state(ComponentTrigger::FaultCompleted)?;
207
208 Ok(())
209 }
210
211 fn reset(&mut self) -> anyhow::Result<()> {
217 self.transition_state(ComponentTrigger::Reset)?; if let Err(e) = self.on_reset() {
220 log_error(self.component_id(), &e);
221 return Err(e); }
223
224 self.transition_state(ComponentTrigger::ResetCompleted)?;
225
226 Ok(())
227 }
228
229 fn dispose(&mut self) -> anyhow::Result<()> {
246 self.transition_state(ComponentTrigger::Dispose)?; let result = self.on_dispose();
249 self.release_subscriptions();
250
251 if let Err(e) = result {
252 log_error(self.component_id(), &e);
253
254 self.transition_state(ComponentTrigger::Fault)?; self.transition_state(ComponentTrigger::FaultCompleted)?; return Err(e);
258 }
259
260 self.transition_state(ComponentTrigger::DisposeCompleted)?;
261
262 Ok(())
263 }
264
265 fn release_subscriptions(&mut self) {}
272
273 fn on_start(&mut self) -> anyhow::Result<()> {
279 log::warn!(
280 "The `on_start` handler was called when not overridden, \
281 it's expected that any actions required when stopping the component \
282 occur here, such as unsubscribing from data",
283 );
284 Ok(())
285 }
286
287 fn on_stop(&mut self) -> anyhow::Result<()> {
293 log::warn!(
294 "The `on_stop` handler was called when not overridden, \
295 it's expected that any actions required when stopping the component \
296 occur here, such as unsubscribing from data",
297 );
298 Ok(())
299 }
300
301 fn on_resume(&mut self) -> anyhow::Result<()> {
307 log::warn!(
308 "The `on_resume` handler was called when not overridden, \
309 it's expected that any actions required when resuming the component \
310 following a stop occur here"
311 );
312 Ok(())
313 }
314
315 fn on_reset(&mut self) -> anyhow::Result<()> {
321 log::warn!(
322 "The `on_reset` handler was called when not overridden, \
323 it's expected that any actions required when resetting the component \
324 occur here, such as resetting indicators and other state"
325 );
326 Ok(())
327 }
328
329 fn on_dispose(&mut self) -> anyhow::Result<()> {
335 Ok(())
336 }
337
338 fn on_degrade(&mut self) -> anyhow::Result<()> {
344 Ok(())
345 }
346
347 fn on_fault(&mut self) -> anyhow::Result<()> {
353 Ok(())
354 }
355}
356
357fn log_error(component: ComponentId, e: &anyhow::Error) {
358 log::error!(component = component.as_str(); "{e}");
359}
360
361#[rustfmt::skip]
362impl ComponentState {
363 pub fn transition(&mut self, trigger: &ComponentTrigger) -> anyhow::Result<Self> {
369 let new_state = match (&self, trigger) {
370 (Self::PreInitialized, ComponentTrigger::Initialize) => Self::Ready,
371 (Self::Ready, ComponentTrigger::Reset) => Self::Resetting,
372 (Self::Ready, ComponentTrigger::Start) => Self::Starting,
373 (Self::Ready, ComponentTrigger::Dispose) => Self::Disposing,
374 (Self::Resetting, ComponentTrigger::ResetCompleted) => Self::Ready,
375 (Self::Starting, ComponentTrigger::StartCompleted) => Self::Running,
376 (Self::Starting, ComponentTrigger::Stop) => Self::Stopping,
377 (Self::Starting, ComponentTrigger::Fault) => Self::Faulting,
378 (Self::Running, ComponentTrigger::Stop) => Self::Stopping,
379 (Self::Running, ComponentTrigger::Degrade) => Self::Degrading,
380 (Self::Running, ComponentTrigger::Fault) => Self::Faulting,
381 (Self::Resuming, ComponentTrigger::Stop) => Self::Stopping,
382 (Self::Resuming, ComponentTrigger::ResumeCompleted) => Self::Running,
383 (Self::Resuming, ComponentTrigger::Fault) => Self::Faulting,
384 (Self::Stopping, ComponentTrigger::StopCompleted) => Self::Stopped,
385 (Self::Stopping, ComponentTrigger::Fault) => Self::Faulting,
386 (Self::Stopped, ComponentTrigger::Reset) => Self::Resetting,
387 (Self::Stopped, ComponentTrigger::Resume) => Self::Resuming,
388 (Self::Stopped, ComponentTrigger::Dispose) => Self::Disposing,
389 (Self::Stopped, ComponentTrigger::Fault) => Self::Faulting,
390 (Self::Degrading, ComponentTrigger::DegradeCompleted) => Self::Degraded,
391 (Self::Degraded, ComponentTrigger::Resume) => Self::Resuming,
392 (Self::Degraded, ComponentTrigger::Stop) => Self::Stopping,
393 (Self::Degraded, ComponentTrigger::Fault) => Self::Faulting,
394 (Self::Disposing, ComponentTrigger::DisposeCompleted) => Self::Disposed,
395 (Self::Disposing, ComponentTrigger::Fault) => Self::Faulting,
396 (Self::Faulting, ComponentTrigger::FaultCompleted) => Self::Faulted,
397 _ => anyhow::bail!("Invalid state trigger {self} -> {trigger}"),
398 };
399 Ok(new_state)
400 }
401}
402
403thread_local! {
404 static COMPONENT_REGISTRY: ComponentRegistry = ComponentRegistry::new();
405}
406
407pub struct ComponentRegistry {
412 components: RefCell<AHashMap<Ustr, Rc<UnsafeCell<dyn Component>>>>,
413 borrows: RefCell<AHashSet<Ustr>>,
414}
415
416impl Debug for ComponentRegistry {
417 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
418 let components_ref = self.components.borrow();
419 let keys: Vec<&Ustr> = components_ref.keys().collect();
420 f.debug_struct(stringify!(ComponentRegistry))
421 .field("components", &keys)
422 .field("active_borrows", &self.borrows.borrow().len())
423 .finish()
424 }
425}
426
427impl Default for ComponentRegistry {
428 fn default() -> Self {
429 Self::new()
430 }
431}
432
433impl ComponentRegistry {
434 pub fn new() -> Self {
435 Self {
436 components: RefCell::new(AHashMap::new()),
437 borrows: RefCell::new(AHashSet::new()),
438 }
439 }
440
441 pub fn insert(&self, id: Ustr, component: Rc<UnsafeCell<dyn Component>>) {
442 self.components.borrow_mut().insert(id, component);
443 }
444
445 pub fn get(&self, id: &Ustr) -> Option<Rc<UnsafeCell<dyn Component>>> {
446 self.components.borrow().get(id).cloned()
447 }
448
449 pub fn remove(&self, id: &Ustr) -> Option<Rc<UnsafeCell<dyn Component>>> {
451 self.components.borrow_mut().remove(id)
452 }
453
454 pub fn is_borrowed(&self, id: &Ustr) -> bool {
456 self.borrows.borrow().contains(id)
457 }
458
459 fn try_borrow(&self, id: Ustr) -> bool {
461 let mut borrows = self.borrows.borrow_mut();
462 if borrows.contains(&id) {
463 false
464 } else {
465 borrows.insert(id);
466 true
467 }
468 }
469
470 fn release_borrow(&self, id: &Ustr) {
472 self.borrows.borrow_mut().remove(id);
473 }
474}
475
476struct BorrowGuard {
481 id: Ustr,
482}
483
484impl BorrowGuard {
485 fn new(id: Ustr) -> Self {
486 Self { id }
487 }
488}
489
490impl Drop for BorrowGuard {
491 fn drop(&mut self) {
492 with_component_registry(|registry| registry.release_borrow(&self.id));
493 }
494}
495
496pub fn with_component_registry<R>(f: impl FnOnce(&ComponentRegistry) -> R) -> R {
497 COMPONENT_REGISTRY.with(f)
498}
499
500pub fn register_component<T>(component: T) -> Rc<UnsafeCell<T>>
502where
503 T: Component + 'static,
504{
505 let component_id = component.component_id().inner();
506 let component_ref = Rc::new(UnsafeCell::new(component));
507
508 let component_trait_ref: Rc<UnsafeCell<dyn Component>> = component_ref.clone();
510 with_component_registry(|registry| registry.insert(component_id, component_trait_ref));
511
512 component_ref
513}
514
515pub fn register_component_actor<T>(component: T) -> Rc<UnsafeCell<T>>
517where
518 T: Component + Actor + 'static,
519{
520 let component_id = component.component_id().inner();
521 let actor_id = component.id();
522 let component_ref = Rc::new(UnsafeCell::new(component));
523
524 let component_trait_ref: Rc<UnsafeCell<dyn Component>> = component_ref.clone();
526 with_component_registry(|registry| registry.insert(component_id, component_trait_ref));
527
528 let actor_trait_ref: Rc<UnsafeCell<dyn Actor>> = component_ref.clone();
530 with_actor_registry(|registry| registry.insert(actor_id, actor_trait_ref));
531
532 component_ref
533}
534
535pub fn start_component(id: &Ustr) -> anyhow::Result<()> {
543 let component_ref = with_component_registry(|registry| {
544 let component_ref = registry
545 .get(id)
546 .ok_or_else(|| anyhow::anyhow!("Component '{id}' not found in global registry"))?;
547
548 if !registry.try_borrow(*id) {
549 anyhow::bail!(
550 "Component '{id}' is already mutably borrowed. \
551 This would create aliasing mutable references (undefined behavior)."
552 );
553 }
554
555 Ok::<_, anyhow::Error>(component_ref)
556 })?;
557
558 let _guard = BorrowGuard::new(*id);
559
560 unsafe {
562 let component = &mut *component_ref.get();
563 component.start()
564 }
565}
566
567pub fn component_state(id: &Ustr) -> anyhow::Result<ComponentState> {
574 let component_ref = with_component_registry(|registry| {
575 let component_ref = registry
576 .get(id)
577 .ok_or_else(|| anyhow::anyhow!("Component '{id}' not found in global registry"))?;
578
579 if !registry.try_borrow(*id) {
580 anyhow::bail!(
581 "Component '{id}' is already mutably borrowed. \
582 This would create aliasing mutable references (undefined behavior)."
583 );
584 }
585
586 Ok::<_, anyhow::Error>(component_ref)
587 })?;
588
589 let _guard = BorrowGuard::new(*id);
590
591 unsafe {
593 let component = &*component_ref.get();
594 Ok(component.state())
595 }
596}
597
598pub fn stop_component(id: &Ustr) -> anyhow::Result<()> {
606 let component_ref = with_component_registry(|registry| {
607 let component_ref = registry
608 .get(id)
609 .ok_or_else(|| anyhow::anyhow!("Component '{id}' not found in global registry"))?;
610
611 if !registry.try_borrow(*id) {
612 anyhow::bail!(
613 "Component '{id}' is already mutably borrowed. \
614 This would create aliasing mutable references (undefined behavior)."
615 );
616 }
617
618 Ok::<_, anyhow::Error>(component_ref)
619 })?;
620
621 let _guard = BorrowGuard::new(*id);
622
623 unsafe {
625 let component = &mut *component_ref.get();
626 component.stop()
627 }
628}
629
630pub fn reset_component(id: &Ustr) -> anyhow::Result<()> {
638 let component_ref = with_component_registry(|registry| {
639 let component_ref = registry
640 .get(id)
641 .ok_or_else(|| anyhow::anyhow!("Component '{id}' not found in global registry"))?;
642
643 if !registry.try_borrow(*id) {
644 anyhow::bail!(
645 "Component '{id}' is already mutably borrowed. \
646 This would create aliasing mutable references (undefined behavior)."
647 );
648 }
649
650 Ok::<_, anyhow::Error>(component_ref)
651 })?;
652
653 let _guard = BorrowGuard::new(*id);
654
655 unsafe {
657 let component = &mut *component_ref.get();
658 component.reset()
659 }
660}
661
662pub fn dispose_component(id: &Ustr) -> anyhow::Result<()> {
670 let component_ref = with_component_registry(|registry| {
671 let component_ref = registry
672 .get(id)
673 .ok_or_else(|| anyhow::anyhow!("Component '{id}' not found in global registry"))?;
674
675 if !registry.try_borrow(*id) {
676 anyhow::bail!(
677 "Component '{id}' is already mutably borrowed. \
678 This would create aliasing mutable references (undefined behavior)."
679 );
680 }
681
682 Ok::<_, anyhow::Error>(component_ref)
683 })?;
684
685 let _guard = BorrowGuard::new(*id);
686
687 unsafe {
689 let component = &mut *component_ref.get();
690 component.dispose()
691 }
692}
693
694pub fn get_component(id: &Ustr) -> Option<Rc<UnsafeCell<dyn Component>>> {
696 with_component_registry(|registry| registry.get(id))
697}
698
699pub fn deregister_component(id: &Ustr) {
704 with_component_registry(|registry| registry.remove(id));
705}
706
707#[cfg(test)]
708pub fn clear_component_registry() {
710 with_component_registry(|registry| {
711 registry.components.borrow_mut().clear();
712 registry.borrows.borrow_mut().clear();
713 });
714}
715
716#[cfg(test)]
717mod tests {
718 use std::{
719 any::Any,
720 sync::atomic::{AtomicBool, Ordering},
721 };
722
723 use rstest::rstest;
724
725 use super::*;
726
727 #[derive(Debug)]
728 struct TestComponent {
729 id: ComponentId,
730 state: ComponentState,
731 should_panic: &'static AtomicBool,
732 }
733
734 impl TestComponent {
735 fn new(name: &str, should_panic: &'static AtomicBool) -> Self {
736 Self {
737 id: ComponentId::new(name),
738 state: ComponentState::Ready,
739 should_panic,
740 }
741 }
742 }
743
744 impl Actor for TestComponent {
745 fn id(&self) -> Ustr {
746 self.id.inner()
747 }
748
749 fn handle(&mut self, _msg: &dyn Any) {}
750
751 fn as_any(&self) -> &dyn Any {
752 self
753 }
754 }
755
756 impl Component for TestComponent {
757 fn component_id(&self) -> ComponentId {
758 self.id
759 }
760
761 fn state(&self) -> ComponentState {
762 self.state
763 }
764
765 fn transition_state(&mut self, trigger: ComponentTrigger) -> anyhow::Result<()> {
766 self.state = self.state.transition(&trigger)?;
767 Ok(())
768 }
769
770 fn register(
771 &mut self,
772 _trader_id: TraderId,
773 _clock: Rc<RefCell<dyn Clock>>,
774 _cache: Rc<RefCell<Cache>>,
775 ) -> anyhow::Result<()> {
776 Ok(())
777 }
778
779 #[expect(clippy::panic_in_result_fn)] fn on_start(&mut self) -> anyhow::Result<()> {
781 assert!(
782 !self.should_panic.load(Ordering::SeqCst),
783 "Intentional panic for testing"
784 );
785 Ok(())
786 }
787 }
788
789 static NO_PANIC: AtomicBool = AtomicBool::new(false);
790 static DO_PANIC: AtomicBool = AtomicBool::new(true);
791
792 #[rstest]
793 fn test_component_borrow_tracking_prevents_double_borrow() {
794 clear_component_registry();
795
796 let id = Ustr::from("test-component-1");
797 let component = TestComponent::new("test-component-1", &NO_PANIC);
798 let component_id = component.id.inner();
799
800 let component_ref = Rc::new(UnsafeCell::new(component));
801 with_component_registry(|registry| registry.insert(component_id, component_ref));
802
803 let result1 = start_component(&id);
805 assert!(result1.is_ok());
806
807 let result2 = stop_component(&id);
809 assert!(result2.is_ok());
810 }
811
812 #[rstest]
813 fn test_component_borrow_released_after_lifecycle_call() {
814 clear_component_registry();
815
816 let id = Ustr::from("test-component-2");
817 let component = TestComponent::new("test-component-2", &NO_PANIC);
818 let component_id = component.id.inner();
819
820 let component_ref = Rc::new(UnsafeCell::new(component));
821 with_component_registry(|registry| registry.insert(component_id, component_ref));
822
823 let _ = start_component(&id);
825
826 assert!(!with_component_registry(
828 |registry| registry.is_borrowed(&id)
829 ));
830 }
831
832 #[rstest]
833 fn test_component_borrow_released_on_panic() {
834 clear_component_registry();
835
836 let id = Ustr::from("test-component-panic");
837 let component = TestComponent::new("test-component-panic", &DO_PANIC);
838 let component_id = component.id.inner();
839
840 let component_ref = Rc::new(UnsafeCell::new(component));
841 with_component_registry(|registry| registry.insert(component_id, component_ref));
842
843 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
845 let _ = start_component(&id);
846 }));
847 assert!(result.is_err(), "Expected panic from on_start");
848
849 assert!(
851 !with_component_registry(|registry| registry.is_borrowed(&id)),
852 "Borrow was not released after panic"
853 );
854 }
855}