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<()> {
190 self.transition_state(ComponentTrigger::Fault)?; if let Err(e) = self.on_fault() {
193 log_error(self.component_id(), &e);
194 return Err(e); }
196
197 self.transition_state(ComponentTrigger::FaultCompleted)?;
198
199 Ok(())
200 }
201
202 fn reset(&mut self) -> anyhow::Result<()> {
208 self.transition_state(ComponentTrigger::Reset)?; if let Err(e) = self.on_reset() {
211 log_error(self.component_id(), &e);
212 return Err(e); }
214
215 self.transition_state(ComponentTrigger::ResetCompleted)?;
216
217 Ok(())
218 }
219
220 fn dispose(&mut self) -> anyhow::Result<()> {
226 self.transition_state(ComponentTrigger::Dispose)?; if let Err(e) = self.on_dispose() {
229 log_error(self.component_id(), &e);
230 return Err(e); }
232
233 self.transition_state(ComponentTrigger::DisposeCompleted)?;
234
235 Ok(())
236 }
237
238 fn on_start(&mut self) -> anyhow::Result<()> {
244 log::warn!(
245 "The `on_start` handler was called when not overridden, \
246 it's expected that any actions required when stopping the component \
247 occur here, such as unsubscribing from data",
248 );
249 Ok(())
250 }
251
252 fn on_stop(&mut self) -> anyhow::Result<()> {
258 log::warn!(
259 "The `on_stop` handler was called when not overridden, \
260 it's expected that any actions required when stopping the component \
261 occur here, such as unsubscribing from data",
262 );
263 Ok(())
264 }
265
266 fn on_resume(&mut self) -> anyhow::Result<()> {
272 log::warn!(
273 "The `on_resume` handler was called when not overridden, \
274 it's expected that any actions required when resuming the component \
275 following a stop occur here"
276 );
277 Ok(())
278 }
279
280 fn on_reset(&mut self) -> anyhow::Result<()> {
286 log::warn!(
287 "The `on_reset` handler was called when not overridden, \
288 it's expected that any actions required when resetting the component \
289 occur here, such as resetting indicators and other state"
290 );
291 Ok(())
292 }
293
294 fn on_dispose(&mut self) -> anyhow::Result<()> {
300 Ok(())
301 }
302
303 fn on_degrade(&mut self) -> anyhow::Result<()> {
309 Ok(())
310 }
311
312 fn on_fault(&mut self) -> anyhow::Result<()> {
318 Ok(())
319 }
320}
321
322fn log_error(component: ComponentId, e: &anyhow::Error) {
323 log::error!(component = component.as_str(); "{e}");
324}
325
326#[rustfmt::skip]
327impl ComponentState {
328 pub fn transition(&mut self, trigger: &ComponentTrigger) -> anyhow::Result<Self> {
334 let new_state = match (&self, trigger) {
335 (Self::PreInitialized, ComponentTrigger::Initialize) => Self::Ready,
336 (Self::Ready, ComponentTrigger::Reset) => Self::Resetting,
337 (Self::Ready, ComponentTrigger::Start) => Self::Starting,
338 (Self::Ready, ComponentTrigger::Dispose) => Self::Disposing,
339 (Self::Resetting, ComponentTrigger::ResetCompleted) => Self::Ready,
340 (Self::Starting, ComponentTrigger::StartCompleted) => Self::Running,
341 (Self::Starting, ComponentTrigger::Stop) => Self::Stopping,
342 (Self::Starting, ComponentTrigger::Fault) => Self::Faulting,
343 (Self::Running, ComponentTrigger::Stop) => Self::Stopping,
344 (Self::Running, ComponentTrigger::Degrade) => Self::Degrading,
345 (Self::Running, ComponentTrigger::Fault) => Self::Faulting,
346 (Self::Resuming, ComponentTrigger::Stop) => Self::Stopping,
347 (Self::Resuming, ComponentTrigger::ResumeCompleted) => Self::Running,
348 (Self::Resuming, ComponentTrigger::Fault) => Self::Faulting,
349 (Self::Stopping, ComponentTrigger::StopCompleted) => Self::Stopped,
350 (Self::Stopping, ComponentTrigger::Fault) => Self::Faulting,
351 (Self::Stopped, ComponentTrigger::Reset) => Self::Resetting,
352 (Self::Stopped, ComponentTrigger::Resume) => Self::Resuming,
353 (Self::Stopped, ComponentTrigger::Dispose) => Self::Disposing,
354 (Self::Stopped, ComponentTrigger::Fault) => Self::Faulting,
355 (Self::Degrading, ComponentTrigger::DegradeCompleted) => Self::Degraded,
356 (Self::Degraded, ComponentTrigger::Resume) => Self::Resuming,
357 (Self::Degraded, ComponentTrigger::Stop) => Self::Stopping,
358 (Self::Degraded, ComponentTrigger::Fault) => Self::Faulting,
359 (Self::Disposing, ComponentTrigger::DisposeCompleted) => Self::Disposed,
360 (Self::Faulting, ComponentTrigger::FaultCompleted) => Self::Faulted,
361 _ => anyhow::bail!("Invalid state trigger {self} -> {trigger}"),
362 };
363 Ok(new_state)
364 }
365}
366
367thread_local! {
368 static COMPONENT_REGISTRY: ComponentRegistry = ComponentRegistry::new();
369}
370
371pub struct ComponentRegistry {
376 components: RefCell<AHashMap<Ustr, Rc<UnsafeCell<dyn Component>>>>,
377 borrows: RefCell<AHashSet<Ustr>>,
378}
379
380impl Debug for ComponentRegistry {
381 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
382 let components_ref = self.components.borrow();
383 let keys: Vec<&Ustr> = components_ref.keys().collect();
384 f.debug_struct(stringify!(ComponentRegistry))
385 .field("components", &keys)
386 .field("active_borrows", &self.borrows.borrow().len())
387 .finish()
388 }
389}
390
391impl Default for ComponentRegistry {
392 fn default() -> Self {
393 Self::new()
394 }
395}
396
397impl ComponentRegistry {
398 pub fn new() -> Self {
399 Self {
400 components: RefCell::new(AHashMap::new()),
401 borrows: RefCell::new(AHashSet::new()),
402 }
403 }
404
405 pub fn insert(&self, id: Ustr, component: Rc<UnsafeCell<dyn Component>>) {
406 self.components.borrow_mut().insert(id, component);
407 }
408
409 pub fn get(&self, id: &Ustr) -> Option<Rc<UnsafeCell<dyn Component>>> {
410 self.components.borrow().get(id).cloned()
411 }
412
413 pub fn is_borrowed(&self, id: &Ustr) -> bool {
415 self.borrows.borrow().contains(id)
416 }
417
418 fn try_borrow(&self, id: Ustr) -> bool {
420 let mut borrows = self.borrows.borrow_mut();
421 if borrows.contains(&id) {
422 false
423 } else {
424 borrows.insert(id);
425 true
426 }
427 }
428
429 fn release_borrow(&self, id: &Ustr) {
431 self.borrows.borrow_mut().remove(id);
432 }
433}
434
435struct BorrowGuard {
440 id: Ustr,
441}
442
443impl BorrowGuard {
444 fn new(id: Ustr) -> Self {
445 Self { id }
446 }
447}
448
449impl Drop for BorrowGuard {
450 fn drop(&mut self) {
451 with_component_registry(|registry| registry.release_borrow(&self.id));
452 }
453}
454
455pub fn with_component_registry<R>(f: impl FnOnce(&ComponentRegistry) -> R) -> R {
456 COMPONENT_REGISTRY.with(f)
457}
458
459pub fn register_component<T>(component: T) -> Rc<UnsafeCell<T>>
461where
462 T: Component + 'static,
463{
464 let component_id = component.component_id().inner();
465 let component_ref = Rc::new(UnsafeCell::new(component));
466
467 let component_trait_ref: Rc<UnsafeCell<dyn Component>> = component_ref.clone();
469 with_component_registry(|registry| registry.insert(component_id, component_trait_ref));
470
471 component_ref
472}
473
474pub fn register_component_actor<T>(component: T) -> Rc<UnsafeCell<T>>
476where
477 T: Component + Actor + 'static,
478{
479 let component_id = component.component_id().inner();
480 let actor_id = component.id();
481 let component_ref = Rc::new(UnsafeCell::new(component));
482
483 let component_trait_ref: Rc<UnsafeCell<dyn Component>> = component_ref.clone();
485 with_component_registry(|registry| registry.insert(component_id, component_trait_ref));
486
487 let actor_trait_ref: Rc<UnsafeCell<dyn Actor>> = component_ref.clone();
489 with_actor_registry(|registry| registry.insert(actor_id, actor_trait_ref));
490
491 component_ref
492}
493
494pub fn start_component(id: &Ustr) -> anyhow::Result<()> {
502 let component_ref = with_component_registry(|registry| {
503 let component_ref = registry
504 .get(id)
505 .ok_or_else(|| anyhow::anyhow!("Component '{id}' not found in global registry"))?;
506
507 if !registry.try_borrow(*id) {
508 anyhow::bail!(
509 "Component '{id}' is already mutably borrowed. \
510 This would create aliasing mutable references (undefined behavior)."
511 );
512 }
513
514 Ok::<_, anyhow::Error>(component_ref)
515 })?;
516
517 let _guard = BorrowGuard::new(*id);
518
519 unsafe {
521 let component = &mut *component_ref.get();
522 component.start()
523 }
524}
525
526pub fn component_state(id: &Ustr) -> anyhow::Result<ComponentState> {
533 let component_ref = with_component_registry(|registry| {
534 let component_ref = registry
535 .get(id)
536 .ok_or_else(|| anyhow::anyhow!("Component '{id}' not found in global registry"))?;
537
538 if !registry.try_borrow(*id) {
539 anyhow::bail!(
540 "Component '{id}' is already mutably borrowed. \
541 This would create aliasing mutable references (undefined behavior)."
542 );
543 }
544
545 Ok::<_, anyhow::Error>(component_ref)
546 })?;
547
548 let _guard = BorrowGuard::new(*id);
549
550 unsafe {
552 let component = &*component_ref.get();
553 Ok(component.state())
554 }
555}
556
557pub fn stop_component(id: &Ustr) -> anyhow::Result<()> {
565 let component_ref = with_component_registry(|registry| {
566 let component_ref = registry
567 .get(id)
568 .ok_or_else(|| anyhow::anyhow!("Component '{id}' not found in global registry"))?;
569
570 if !registry.try_borrow(*id) {
571 anyhow::bail!(
572 "Component '{id}' is already mutably borrowed. \
573 This would create aliasing mutable references (undefined behavior)."
574 );
575 }
576
577 Ok::<_, anyhow::Error>(component_ref)
578 })?;
579
580 let _guard = BorrowGuard::new(*id);
581
582 unsafe {
584 let component = &mut *component_ref.get();
585 component.stop()
586 }
587}
588
589pub fn reset_component(id: &Ustr) -> anyhow::Result<()> {
597 let component_ref = with_component_registry(|registry| {
598 let component_ref = registry
599 .get(id)
600 .ok_or_else(|| anyhow::anyhow!("Component '{id}' not found in global registry"))?;
601
602 if !registry.try_borrow(*id) {
603 anyhow::bail!(
604 "Component '{id}' is already mutably borrowed. \
605 This would create aliasing mutable references (undefined behavior)."
606 );
607 }
608
609 Ok::<_, anyhow::Error>(component_ref)
610 })?;
611
612 let _guard = BorrowGuard::new(*id);
613
614 unsafe {
616 let component = &mut *component_ref.get();
617 component.reset()
618 }
619}
620
621pub fn dispose_component(id: &Ustr) -> anyhow::Result<()> {
629 let component_ref = with_component_registry(|registry| {
630 let component_ref = registry
631 .get(id)
632 .ok_or_else(|| anyhow::anyhow!("Component '{id}' not found in global registry"))?;
633
634 if !registry.try_borrow(*id) {
635 anyhow::bail!(
636 "Component '{id}' is already mutably borrowed. \
637 This would create aliasing mutable references (undefined behavior)."
638 );
639 }
640
641 Ok::<_, anyhow::Error>(component_ref)
642 })?;
643
644 let _guard = BorrowGuard::new(*id);
645
646 unsafe {
648 let component = &mut *component_ref.get();
649 component.dispose()
650 }
651}
652
653pub fn get_component(id: &Ustr) -> Option<Rc<UnsafeCell<dyn Component>>> {
655 with_component_registry(|registry| registry.get(id))
656}
657
658#[cfg(test)]
659pub fn clear_component_registry() {
661 with_component_registry(|registry| {
662 registry.components.borrow_mut().clear();
663 registry.borrows.borrow_mut().clear();
664 });
665}
666
667#[cfg(test)]
668mod tests {
669 use std::{
670 any::Any,
671 sync::atomic::{AtomicBool, Ordering},
672 };
673
674 use rstest::rstest;
675
676 use super::*;
677
678 #[derive(Debug)]
679 struct TestComponent {
680 id: ComponentId,
681 state: ComponentState,
682 should_panic: &'static AtomicBool,
683 }
684
685 impl TestComponent {
686 fn new(name: &str, should_panic: &'static AtomicBool) -> Self {
687 Self {
688 id: ComponentId::new(name),
689 state: ComponentState::Ready,
690 should_panic,
691 }
692 }
693 }
694
695 impl Actor for TestComponent {
696 fn id(&self) -> Ustr {
697 self.id.inner()
698 }
699
700 fn handle(&mut self, _msg: &dyn Any) {}
701
702 fn as_any(&self) -> &dyn Any {
703 self
704 }
705 }
706
707 impl Component for TestComponent {
708 fn component_id(&self) -> ComponentId {
709 self.id
710 }
711
712 fn state(&self) -> ComponentState {
713 self.state
714 }
715
716 fn transition_state(&mut self, trigger: ComponentTrigger) -> anyhow::Result<()> {
717 self.state = self.state.transition(&trigger)?;
718 Ok(())
719 }
720
721 fn register(
722 &mut self,
723 _trader_id: TraderId,
724 _clock: Rc<RefCell<dyn Clock>>,
725 _cache: Rc<RefCell<Cache>>,
726 ) -> anyhow::Result<()> {
727 Ok(())
728 }
729
730 #[expect(clippy::panic_in_result_fn)] fn on_start(&mut self) -> anyhow::Result<()> {
732 assert!(
733 !self.should_panic.load(Ordering::SeqCst),
734 "Intentional panic for testing"
735 );
736 Ok(())
737 }
738 }
739
740 static NO_PANIC: AtomicBool = AtomicBool::new(false);
741 static DO_PANIC: AtomicBool = AtomicBool::new(true);
742
743 #[rstest]
744 fn test_component_borrow_tracking_prevents_double_borrow() {
745 clear_component_registry();
746
747 let id = Ustr::from("test-component-1");
748 let component = TestComponent::new("test-component-1", &NO_PANIC);
749 let component_id = component.id.inner();
750
751 let component_ref = Rc::new(UnsafeCell::new(component));
752 with_component_registry(|registry| registry.insert(component_id, component_ref));
753
754 let result1 = start_component(&id);
756 assert!(result1.is_ok());
757
758 let result2 = stop_component(&id);
760 assert!(result2.is_ok());
761 }
762
763 #[rstest]
764 fn test_component_borrow_released_after_lifecycle_call() {
765 clear_component_registry();
766
767 let id = Ustr::from("test-component-2");
768 let component = TestComponent::new("test-component-2", &NO_PANIC);
769 let component_id = component.id.inner();
770
771 let component_ref = Rc::new(UnsafeCell::new(component));
772 with_component_registry(|registry| registry.insert(component_id, component_ref));
773
774 let _ = start_component(&id);
776
777 assert!(!with_component_registry(
779 |registry| registry.is_borrowed(&id)
780 ));
781 }
782
783 #[rstest]
784 fn test_component_borrow_released_on_panic() {
785 clear_component_registry();
786
787 let id = Ustr::from("test-component-panic");
788 let component = TestComponent::new("test-component-panic", &DO_PANIC);
789 let component_id = component.id.inner();
790
791 let component_ref = Rc::new(UnsafeCell::new(component));
792 with_component_registry(|registry| registry.insert(component_id, component_ref));
793
794 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
796 let _ = start_component(&id);
797 }));
798 assert!(result.is_err(), "Expected panic from on_start");
799
800 assert!(
802 !with_component_registry(|registry| registry.is_borrowed(&id)),
803 "Borrow was not released after panic"
804 );
805 }
806}