nautilus_common/actor/
registry.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
16//! Thread-local actor registry with access guards.
17//!
18//! # Design
19//!
20//! The actor registry stores actors in thread-local storage and provides access via
21//! [`ActorRef<T>`] guards. This design addresses several constraints:
22//!
23//! - **Use-after-free prevention**: `ActorRef` holds an `Rc` clone, keeping the actor
24//!   alive even if removed from the registry while the guard exists.
25//! - **Scoped registry access**: Registry access stays tied to the thread-local storage
26//!   access callback.
27//! - **Thread-local only**: Guards must not be sent across threads.
28//!
29//! # Limitations
30//!
31//! - **Aliasing not prevented**: Two guards can exist for the same actor simultaneously,
32//!   allowing aliased mutable access. This is undefined behavior if both guards create
33//!   overlapping references to the same actor. The current actor dispatch model relies
34//!   on same-actor re-entrant lookups, so fixing this requires a broader dispatch and
35//!   ownership redesign.
36//!
37//! # Invariants
38//!
39//! These contracts must hold regardless of how the registry is implemented
40//! internally. The first three are verified by tests in this module. The
41//! fourth is a usage discipline enforced by convention.
42//!
43//! - **Thread isolation**: Each thread has its own registry instance. An actor
44//!   registered on one thread is never visible from another.
45//! - **Guard survival**: An [`ActorRef`] keeps its actor alive via reference
46//!   counting. Removing or replacing an actor in the registry does not invalidate
47//!   existing guards.
48//! - **Type safety**: [`get_actor_unchecked`] and [`try_get_actor_unchecked`]
49//!   verify the concrete type at runtime before casting. A type mismatch panics
50//!   or returns `None`, respectively.
51//! - **Short-lived guards**: Guards must be obtained, used, and dropped within a
52//!   single synchronous scope. Never store an [`ActorRef`] in a struct or hold
53//!   one across an `.await` point.
54
55use std::{
56    any::TypeId,
57    cell::{RefCell, UnsafeCell},
58    fmt::Debug,
59    marker::PhantomData,
60    ops::{Deref, DerefMut},
61    rc::Rc,
62};
63
64use ahash::AHashMap;
65use ustr::Ustr;
66
67use super::Actor;
68
69/// A guard providing mutable access to an actor.
70///
71/// This guard holds an `Rc` reference to keep the actor alive.
72pub struct ActorRef<T: Actor> {
73    actor_rc: Rc<UnsafeCell<dyn Actor>>,
74    _marker: PhantomData<T>,
75}
76
77impl<T: Actor> Debug for ActorRef<T> {
78    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
79        f.debug_struct(stringify!(ActorRef))
80            .field("actor_id", &self.deref().id())
81            .finish()
82    }
83}
84
85impl<T: Actor> Deref for ActorRef<T> {
86    type Target = T;
87
88    fn deref(&self) -> &Self::Target {
89        // SAFETY: Type was verified at construction time.
90        unsafe { &*(self.actor_rc.get() as *const T) }
91    }
92}
93
94impl<T: Actor> DerefMut for ActorRef<T> {
95    fn deref_mut(&mut self) -> &mut Self::Target {
96        // SAFETY: Type was verified at construction time.
97        unsafe { &mut *self.actor_rc.get().cast::<T>() }
98    }
99}
100
101thread_local! {
102    static ACTOR_REGISTRY: ActorRegistry = ActorRegistry::new();
103}
104
105/// Registry for storing actors.
106pub struct ActorRegistry {
107    actors: RefCell<AHashMap<Ustr, Registration>>,
108}
109
110impl Debug for ActorRegistry {
111    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
112        let actors_ref = self.actors.borrow();
113        let keys: Vec<&Ustr> = actors_ref.keys().collect();
114        f.debug_struct(stringify!(ActorRegistry))
115            .field("actors", &keys)
116            .finish()
117    }
118}
119
120impl Default for ActorRegistry {
121    fn default() -> Self {
122        Self::new()
123    }
124}
125
126impl ActorRegistry {
127    pub fn new() -> Self {
128        Self {
129            actors: RefCell::new(AHashMap::new()),
130        }
131    }
132
133    pub fn insert(&self, id: Ustr, actor: Rc<UnsafeCell<dyn Actor>>) {
134        let mut actors = self.actors.borrow_mut();
135        if actors.contains_key(&id) {
136            log::warn!("Replacing existing actor with id: {id}");
137        }
138        let previous = actors.insert(
139            id,
140            Registration {
141                actor,
142                identity: Rc::new(()),
143            },
144        );
145        drop(actors);
146        drop(previous);
147    }
148
149    pub fn get(&self, id: &Ustr) -> Option<Rc<UnsafeCell<dyn Actor>>> {
150        self.actors
151            .borrow()
152            .get(id)
153            .map(|entry| entry.actor.clone())
154    }
155
156    /// Returns the number of registered actors.
157    pub fn len(&self) -> usize {
158        self.actors.borrow().len()
159    }
160
161    /// Checks if the registry is empty.
162    pub fn is_empty(&self) -> bool {
163        self.actors.borrow().is_empty()
164    }
165
166    /// Removes an actor from the registry.
167    pub fn remove(&self, id: &Ustr) -> Option<Rc<UnsafeCell<dyn Actor>>> {
168        self.actors.borrow_mut().remove(id).map(|entry| entry.actor)
169    }
170
171    /// Checks if an actor with the `id` exists.
172    pub fn contains(&self, id: &Ustr) -> bool {
173        self.actors.borrow().contains_key(id)
174    }
175}
176
177pub fn with_actor_registry<R>(f: impl FnOnce(&ActorRegistry) -> R) -> R {
178    ACTOR_REGISTRY.with(f)
179}
180
181/// Registers an actor.
182pub fn register_actor<T>(actor: T) -> Rc<UnsafeCell<T>>
183where
184    T: Actor + 'static,
185{
186    let actor_id = actor.id();
187    let actor_ref = Rc::new(UnsafeCell::new(actor));
188
189    // Register as Actor (message handling only)
190    let actor_trait_ref: Rc<UnsafeCell<dyn Actor>> = actor_ref.clone();
191    with_actor_registry(|registry| registry.insert(actor_id, actor_trait_ref));
192
193    actor_ref
194}
195
196pub fn get_actor(id: &Ustr) -> Option<Rc<UnsafeCell<dyn Actor>>> {
197    with_actor_registry(|registry| registry.get(id))
198}
199
200/// Removes the actor with `id` from the registry.
201///
202/// Only the exact ID is removed, so unrelated actors sharing the thread-local registry are
203/// untouched.
204pub fn deregister_actor(id: &Ustr) {
205    with_actor_registry(|registry| registry.remove(id));
206}
207
208/// Returns a guard providing mutable access to the registered actor of type `T`.
209///
210/// The returned [`ActorRef`] holds an `Rc` to keep the actor alive, preventing
211/// use-after-free if the actor is removed from the registry.
212///
213/// # Panics
214///
215/// - Panics if no actor with the specified `id` is found in the registry.
216/// - Panics if the stored actor is not of type `T`.
217#[must_use]
218pub fn get_actor_unchecked<T: Actor>(id: &Ustr) -> ActorRef<T> {
219    let actor_rc = with_actor_registry(|registry| registry.get(id))
220        .unwrap_or_else(|| panic!("Actor for {id} not found"));
221
222    match actor_ref_from_rc(actor_rc) {
223        Ok(actor_ref) => actor_ref,
224        Err(ActorRefError {
225            expected_type,
226            actual_type,
227        }) => {
228            panic!(
229                "Actor type mismatch for '{id}': expected {expected_type:?}, found {actual_type:?}"
230            )
231        }
232    }
233}
234
235/// Attempts to get a guard providing mutable access to the registered actor.
236///
237/// Returns `None` if the actor is not found or the type doesn't match.
238#[must_use]
239pub fn try_get_actor_unchecked<T: Actor>(id: &Ustr) -> Option<ActorRef<T>> {
240    let actor_rc = with_actor_registry(|registry| registry.get(id))?;
241    actor_ref_from_rc(actor_rc).ok()
242}
243
244#[derive(Debug)]
245struct ActorRefError {
246    expected_type: TypeId,
247    actual_type: TypeId,
248}
249
250fn actor_ref_from_rc<T: Actor>(
251    actor_rc: Rc<UnsafeCell<dyn Actor>>,
252) -> Result<ActorRef<T>, ActorRefError> {
253    // SAFETY: Get a reference to check the type before casting.
254    let actor_ref = unsafe { &*actor_rc.get() };
255    let actual_type = actor_ref.as_any().type_id();
256    let expected_type = TypeId::of::<T>();
257
258    if actual_type != expected_type {
259        return Err(ActorRefError {
260            expected_type,
261            actual_type,
262        });
263    }
264
265    Ok(ActorRef {
266        actor_rc,
267        _marker: PhantomData,
268    })
269}
270
271/// Checks if an actor with the `id` exists in the registry.
272pub fn actor_exists(id: &Ustr) -> bool {
273    with_actor_registry(|registry| registry.contains(id))
274}
275
276/// Returns the number of registered actors.
277pub fn actor_count() -> usize {
278    with_actor_registry(ActorRegistry::len)
279}
280
281#[derive(Clone)]
282struct Registration {
283    actor: Rc<UnsafeCell<dyn Actor>>,
284    identity: Rc<()>,
285}
286
287#[allow(
288    dead_code,
289    reason = "registration-bound admission remains inactive until runtime integration"
290)]
291pub(super) fn reserve_actor<T: Actor, E: 'static>(
292    id: Ustr,
293    heap_bytes: usize,
294) -> Option<ActorAdmission<T, E>> {
295    let registration = ACTOR_REGISTRY
296        .try_with(|registry| registry.actors.borrow().get(&id).cloned())
297        .ok()
298        .flatten()?;
299    let admission = super::dispatch::reserve(heap_bytes)?;
300    Some(ActorAdmission {
301        id,
302        registration,
303        admission,
304    })
305}
306
307#[allow(
308    dead_code,
309    reason = "registration-bound admission remains inactive until runtime integration"
310)]
311pub(super) struct ActorAdmission<T: Actor, E> {
312    id: Ustr,
313    registration: Registration,
314    admission: super::dispatch::Admission<ActorDelivery<T, E>>,
315}
316
317#[allow(
318    dead_code,
319    reason = "registration-bound admission remains inactive until runtime integration"
320)]
321impl<T: Actor, E: 'static> ActorAdmission<T, E> {
322    pub(super) fn commit(self, event: E, handler: fn(&mut T, &E)) {
323        self.admission.commit(
324            ActorDelivery {
325                id: self.id,
326                registration: self.registration,
327                event,
328                handler,
329            },
330            ActorDelivery::run,
331        );
332    }
333}
334
335struct ActorDelivery<T: Actor, E> {
336    id: Ustr,
337    registration: Registration,
338    event: E,
339    handler: fn(&mut T, &E),
340}
341
342impl<T: Actor, E> ActorDelivery<T, E> {
343    fn run(&mut self) -> bool {
344        let current =
345            ACTOR_REGISTRY
346                .try_with(|registry| {
347                    registry.actors.borrow().get(&self.id).is_some_and(|entry| {
348                        Rc::ptr_eq(&entry.identity, &self.registration.identity)
349                    })
350                })
351                .unwrap_or(false);
352
353        if !current {
354            return true;
355        }
356
357        match super::access::ActorGuard::acquire(self.registration.actor.clone()) {
358            Ok(mut actor) => {
359                (self.handler)(&mut actor, &self.event);
360                true
361            }
362            Err(super::access::ActorAccessError::Busy) => false,
363            Err(_) => {
364                super::dispatch::record_failure(super::dispatch::DispatchError::InvalidDestination);
365                true
366            }
367        }
368    }
369}
370
371#[cfg(test)]
372/// Clears the actor registry (for test isolation).
373pub fn clear_actor_registry() {
374    let actors = with_actor_registry(|registry| std::mem::take(&mut *registry.actors.borrow_mut()));
375    drop(actors);
376}
377
378#[cfg(test)]
379mod tests {
380    use std::any::Any;
381
382    use rstest::rstest;
383
384    use super::*;
385
386    #[derive(Debug)]
387    struct TestActor {
388        id: Ustr,
389        value: i32,
390    }
391
392    impl Actor for TestActor {
393        fn id(&self) -> Ustr {
394            self.id
395        }
396        fn handle(&mut self, _msg: &dyn Any) {}
397        fn as_any(&self) -> &dyn Any {
398            self
399        }
400    }
401
402    #[rstest]
403    fn owned_delivery_waits_for_access_and_cancels_same_allocation_registration() {
404        super::super::dispatch::clear().unwrap();
405        clear_actor_registry();
406        let id = Ustr::from("owned-registration");
407        let allocation = register_actor(TestActor { id, value: 11 });
408        let guard =
409            super::super::access::ActorGuard::<TestActor>::acquire(allocation.clone()).unwrap();
410        reserve_actor::<TestActor, _>(id, 0)
411            .unwrap()
412            .commit(23, |actor, value| actor.value = *value);
413        assert_eq!(super::super::dispatch::drain(1).unwrap().delivered, 0);
414        drop(guard);
415        assert_eq!(super::super::dispatch::drain(1).unwrap().delivered, 1);
416        assert_eq!(get_actor_unchecked::<TestActor>(&id).value, 23);
417        reserve_actor::<TestActor, _>(id, 0)
418            .unwrap()
419            .commit(37, |actor, value| actor.value = *value);
420        deregister_actor(&id);
421        with_actor_registry(|registry| registry.insert(id, allocation));
422        assert_eq!(super::super::dispatch::drain(1).unwrap().delivered, 1);
423        assert_eq!(get_actor_unchecked::<TestActor>(&id).value, 23);
424        super::super::dispatch::clear().unwrap();
425    }
426
427    #[rstest]
428    fn owned_delivery_retries_busy_allocation() {
429        super::super::dispatch::clear().unwrap();
430        clear_actor_registry();
431        let id = Ustr::from("busy-delivery");
432        let allocation = register_actor(TestActor { id, value: 11 });
433        let registration = with_actor_registry(|registry| registry.actors.borrow()[&id].clone());
434        let mut delivery = ActorDelivery {
435            id,
436            registration,
437            event: 23,
438            handler: |actor: &mut TestActor, value: &i32| actor.value = *value,
439        };
440        let guard = super::super::access::ActorGuard::<TestActor>::acquire(allocation).unwrap();
441        let busy = delivery.run();
442        assert!(!busy);
443        assert_eq!(guard.value, 11);
444        assert_eq!(super::super::dispatch::failure(), None);
445        drop(guard);
446        let delivered = delivery.run();
447        assert!(delivered);
448        assert_eq!(get_actor_unchecked::<TestActor>(&id).value, 23);
449        assert_eq!(super::super::dispatch::failure(), None);
450        clear_actor_registry();
451    }
452
453    #[rstest]
454    fn owned_delivery_rejects_wrong_actor_type() {
455        #[derive(Debug)]
456        struct OtherActor;
457
458        impl Actor for OtherActor {
459            fn id(&self) -> Ustr {
460                Ustr::from("wrong-delivery-type")
461            }
462            fn handle(&mut self, _msg: &dyn Any) {}
463            fn as_any(&self) -> &dyn Any {
464                self
465            }
466        }
467
468        super::super::dispatch::clear().unwrap();
469        clear_actor_registry();
470        let id = OtherActor.id();
471        register_actor(OtherActor);
472        reserve_actor::<TestActor, _>(id, 0)
473            .unwrap()
474            .commit(23, |_, _| panic!("wrong-type handler must not run"));
475        let result = super::super::dispatch::drain(1);
476        assert_eq!(
477            result,
478            Err(super::super::dispatch::DispatchError::InvalidDestination)
479        );
480        assert_eq!(
481            super::super::dispatch::failure(),
482            Some(super::super::dispatch::DispatchError::InvalidDestination)
483        );
484        super::super::dispatch::clear().unwrap();
485        clear_actor_registry();
486    }
487
488    #[rstest]
489    fn actor_reservation_rejects_after_registry_teardown() {
490        struct Probe;
491
492        impl Drop for Probe {
493            fn drop(&mut self) {
494                assert!(ACTOR_REGISTRY.try_with(|_| ()).is_err());
495                assert!(reserve_actor::<TestActor, ()>(Ustr::from("teardown"), 0).is_none());
496            }
497        }
498
499        thread_local! {
500            static PROBE: Probe = const { Probe };
501        }
502
503        std::thread::spawn(|| {
504            PROBE.with(|_| ());
505            register_actor(TestActor {
506                id: Ustr::from("teardown"),
507                value: 11,
508            });
509        })
510        .join()
511        .unwrap();
512    }
513
514    #[rstest]
515    fn test_register_and_get_actor() {
516        clear_actor_registry();
517
518        let id = Ustr::from("test-actor");
519        let actor = TestActor { id, value: 42 };
520        register_actor(actor);
521
522        let actor_ref = get_actor_unchecked::<TestActor>(&id);
523        assert_eq!(actor_ref.value, 42);
524    }
525
526    #[rstest]
527    fn test_mutation_through_reference() {
528        clear_actor_registry();
529
530        let id = Ustr::from("test-actor-mut");
531        let actor = TestActor { id, value: 0 };
532        register_actor(actor);
533
534        let mut actor_ref = get_actor_unchecked::<TestActor>(&id);
535        actor_ref.value = 999;
536        drop(actor_ref);
537
538        let actor_ref2 = get_actor_unchecked::<TestActor>(&id);
539        assert_eq!(actor_ref2.value, 999);
540    }
541
542    #[rstest]
543    fn test_try_get_returns_none_for_missing() {
544        clear_actor_registry();
545
546        let id = Ustr::from("nonexistent");
547        let result = try_get_actor_unchecked::<TestActor>(&id);
548        assert!(result.is_none());
549    }
550
551    #[rstest]
552    fn test_try_get_returns_none_for_wrong_type() {
553        #[derive(Debug)]
554        struct OtherActor {
555            id: Ustr,
556        }
557
558        impl Actor for OtherActor {
559            fn id(&self) -> Ustr {
560                self.id
561            }
562            fn handle(&mut self, _msg: &dyn Any) {}
563            fn as_any(&self) -> &dyn Any {
564                self
565            }
566        }
567
568        clear_actor_registry();
569
570        let id = Ustr::from("other-actor");
571        let actor = OtherActor { id };
572        register_actor(actor);
573
574        let result = try_get_actor_unchecked::<TestActor>(&id);
575        assert!(result.is_none());
576    }
577
578    #[rstest]
579    fn test_registry_is_thread_local() {
580        clear_actor_registry();
581
582        let id = Ustr::from("thread-local-actor");
583        let actor = TestActor { id, value: 42 };
584        register_actor(actor);
585
586        assert!(actor_exists(&id));
587        assert_eq!(actor_count(), 1);
588
589        let visible_on_other_thread = std::thread::spawn(move || {
590            // Each thread gets its own empty registry
591            (actor_exists(&id), actor_count())
592        })
593        .join()
594        .unwrap();
595
596        assert!(!visible_on_other_thread.0);
597        assert_eq!(visible_on_other_thread.1, 0);
598    }
599
600    #[rstest]
601    fn test_actor_ref_survives_registry_removal() {
602        clear_actor_registry();
603
604        let id = Ustr::from("removable-actor");
605        let actor = TestActor { id, value: 7 };
606        register_actor(actor);
607        assert_eq!(actor_count(), 1);
608
609        let mut guard = get_actor_unchecked::<TestActor>(&id);
610
611        with_actor_registry(|registry| {
612            registry.remove(&id);
613        });
614        assert!(!actor_exists(&id));
615        assert_eq!(actor_count(), 0);
616
617        assert_eq!(guard.value, 7);
618        guard.value = 99;
619        assert_eq!(guard.value, 99);
620    }
621
622    #[rstest]
623    fn test_deregister_actor_removes_only_requested_actor_and_retains_guard() {
624        clear_actor_registry();
625
626        let removed_id = Ustr::from("removed-actor");
627        let retained_id = Ustr::from("retained-actor");
628        register_actor(TestActor {
629            id: removed_id,
630            value: 7,
631        });
632        register_actor(TestActor {
633            id: retained_id,
634            value: 11,
635        });
636        let removed_guard = get_actor_unchecked::<TestActor>(&removed_id);
637
638        deregister_actor(&removed_id);
639
640        assert!(!actor_exists(&removed_id));
641        assert!(actor_exists(&retained_id));
642        assert_eq!(actor_count(), 1);
643        assert_eq!(removed_guard.value, 7);
644        assert_eq!(get_actor_unchecked::<TestActor>(&retained_id).value, 11);
645    }
646
647    #[rstest]
648    fn test_actor_ref_survives_same_id_replacement() {
649        clear_actor_registry();
650
651        let id = Ustr::from("replaceable-actor");
652        let actor_a = TestActor { id, value: 1 };
653        register_actor(actor_a);
654
655        let guard_a = get_actor_unchecked::<TestActor>(&id);
656        assert_eq!(guard_a.value, 1);
657
658        let actor_b = TestActor { id, value: 2 };
659        register_actor(actor_b);
660
661        // Old guard still sees actor A
662        assert_eq!(guard_a.value, 1);
663
664        // Fresh lookup sees actor B
665        let guard_b = get_actor_unchecked::<TestActor>(&id);
666        assert_eq!(guard_b.value, 2);
667        assert_eq!(actor_count(), 1);
668    }
669
670    #[should_panic(expected = "Actor type mismatch")]
671    #[rstest]
672    fn test_get_actor_unchecked_panics_on_type_mismatch() {
673        #[derive(Debug)]
674        struct OtherActor {
675            id: Ustr,
676        }
677
678        impl Actor for OtherActor {
679            fn id(&self) -> Ustr {
680                self.id
681            }
682            fn handle(&mut self, _msg: &dyn Any) {}
683            fn as_any(&self) -> &dyn Any {
684                self
685            }
686        }
687
688        clear_actor_registry();
689
690        let id = Ustr::from("typed-actor");
691        let actor = OtherActor { id };
692        register_actor(actor);
693
694        let _guard = get_actor_unchecked::<TestActor>(&id);
695    }
696}
nautilus_common/actor/registry.rs

nautilus_common/actor/
registry.rs
