Skip to main content

bombay/runtime/
system.rs

1//! Actor execution ownership.
2
3use behavior::{Address, Behavior, BirthMode, Never, ShutdownEvent, TimeEvent};
4use bombay_engine::{Driver, Environment, RunError, RunExit};
5use observe::ObservationSpace;
6use std::sync::Arc;
7use std::sync::atomic::AtomicBool;
8
9use super::lifecycle::{IncarnationReporter, LifecycleFactory};
10use super::{
11    CreationFailure, LaunchMode, Lifecycle, NoLifecycle, PreparedIncarnation,
12    ProvisionalIncarnation,
13};
14use super::{IncarnationEffects, NoParent, ParentReporter};
15use crate::{
16    ActorEnvironment, ActorRef, ChildLease, ChildRuntime, EndpointRegistry, EventSender, Handle,
17    IncarnationEndpoint, MailboxAnchor, MailboxConfig, MailboxDeliveryClosed, MailboxReceiver,
18    MailboxSender, ObservesCreations, RejectedDelivery, RouteSends, RuntimeBirthMode,
19    ShutdownRequestError, SystemChildren, TaskOutcome,
20};
21
22/// The result produced by one actor task.
23#[doc(hidden)]
24pub(crate) type ActorResult<B, E> = Result<
25    RunExit<behavior::Exit<<B as Behavior>::Addr>>,
26    RunError<<B as Behavior>::Error, <E as Environment>::Error>,
27>;
28
29/// Tokio-backed actor execution boundary with shared typed routing.
30pub struct System<R, L = NoLifecycle> {
31    mailbox: MailboxConfig,
32    router: R,
33    lifecycle: L,
34}
35
36impl<R: Clone, L: Clone> Clone for System<R, L> {
37    fn clone(&self) -> Self {
38        Self {
39            mailbox: self.mailbox,
40            router: self.router.clone(),
41            lifecycle: self.lifecycle.clone(),
42        }
43    }
44}
45
46impl<R> System<R, NoLifecycle> {
47    /// Construct a Tokio-backed system from its communication configuration and router.
48    pub const fn new(mailbox: MailboxConfig, router: R) -> Self {
49        Self {
50            mailbox,
51            router,
52            lifecycle: NoLifecycle,
53        }
54    }
55
56    /// Construct a Tokio-backed system with a statically dispatched lifecycle sink.
57    pub const fn with_lifecycle<S>(
58        mailbox: MailboxConfig,
59        router: R,
60        lifecycle: S,
61    ) -> System<R, Lifecycle<S>> {
62        System {
63            mailbox,
64            router,
65            lifecycle: Lifecycle(lifecycle),
66        }
67    }
68}
69
70/// Address projected from a behavior.
71type AddrOf<B> = <B as Behavior>::Addr;
72
73/// Event projected from a behavior.
74type EventOf<B> = <B as Behavior>::Event;
75
76/// Nonce projected from a behavior's address.
77type NonceOf<B> = <AddrOf<B> as Address>::Nonce;
78
79/// Non-owning mailbox endpoint for a behavior's event protocol.
80type AnchorOf<B> = MailboxAnchor<EventOf<B>>;
81
82/// Child behavior projected from a behavior's birth mode.
83type ChildOf<B> = <<B as Behavior>::Birth as BirthMode>::Child;
84
85/// Child runtime constructed by a behavior's birth mode under `Y`.
86type RuntimeOf<B, Y> =
87    <<B as Behavior>::Birth as RuntimeBirthMode<AddrOf<B>, Y, AnchorOf<B>>>::Runtime;
88
89/// Registered endpoint published for one behavior incarnation.
90type AnchorEndpoint<B> = IncarnationEndpoint<AddrOf<B>, ActorRef<AddrOf<B>, AnchorOf<B>>>;
91
92/// Generation-local effects interpreting one behavior's send algebra.
93type EffectsOf<R, B, Y, P> = IncarnationEffects<
94    R,
95    NonceOf<B>,
96    <RuntimeOf<B, Y> as ChildRuntime<AddrOf<B>, ChildOf<B>, AnchorOf<B>>>::Lease,
97    AnchorOf<B>,
98    P,
99    AddrOf<B>,
100>;
101
102/// Direct reference produced for a behavior by Bombay Communication.
103#[doc(hidden)]
104pub(crate) type BehaviorRef<B, L = NoLifecycle> = ActorRef<AddrOf<B>, MailboxSender<EventOf<B>>, L>;
105
106/// Child runtime derived from a behavior's birth mode.
107#[doc(hidden)]
108pub(crate) type BehaviorChildren<R, B, L = NoLifecycle> = RuntimeOf<B, System<R, L>>;
109
110/// Environment constructed for a behavior by a configured system.
111#[doc(hidden)]
112pub(crate) type BehaviorEnvironment<R, B, P = NoParent, L = NoLifecycle> =
113    ActorEnvironment<B, MailboxReceiver<EventOf<B>>, R, BehaviorChildren<R, B, L>, AnchorOf<B>, P>;
114
115/// Registration ownership token claimed for one behavior incarnation.
116#[doc(hidden)]
117type BehaviorRegistration<R, B> =
118    <R as EndpointRegistry<AddrOf<B>, <B as Behavior>::Msg, AnchorEndpoint<B>>>::Registration;
119
120/// Lifecycle reporter derived for one behavior incarnation.
121type BehaviorReporter<R, B, L> =
122    <L as LifecycleFactory<AddrOf<B>, BehaviorRegistration<R, B>>>::Reporter;
123
124/// Exact terminal value produced by one transactionally activated root.
125#[doc(hidden)]
126pub type RootOutcome<R, B, L = NoLifecycle> =
127    ActorResult<B, BehaviorEnvironment<R, B, NoParent, L>>;
128
129/// Cloneable, delivery-only capability for one activated root incarnation.
130pub struct RootEndpoint<B: Behavior<Ph = Never>> {
131    inner: ActorRef<AddrOf<B>, AnchorOf<B>>,
132}
133
134impl<B> Clone for RootEndpoint<B>
135where
136    B: Behavior<Ph = Never>,
137    AddrOf<B>: Clone,
138    AnchorOf<B>: Clone,
139{
140    fn clone(&self) -> Self {
141        Self {
142            inner: self.inner.clone(),
143        }
144    }
145}
146
147impl<B> crate::DeliveryEndpoint<AddrOf<B>, B::Msg> for RootEndpoint<B>
148where
149    B: Behavior<Ph = Never>,
150    AddrOf<B>: Send + Sync,
151    B::Event: behavior::UserEvent<Addr = AddrOf<B>> + Send,
152    B::Msg: Send,
153{
154    type Error = MailboxDeliveryClosed;
155
156    async fn deliver(
157        &self,
158        from: AddrOf<B>,
159        message: B::Msg,
160    ) -> Result<(), RejectedDelivery<B::Msg, Self::Error>> {
161        crate::DeliveryEndpoint::deliver(&self.inner, from, message).await
162    }
163}
164
165/// Affine retirement authority for one activated root.
166pub struct RootRetirement<R, T> {
167    handle: Handle<R, T>,
168}
169
170impl<R, T> RootRetirement<R, T> {
171    /// Request hard cancellation of this exact incarnation.
172    pub fn abort(&self) {
173        self.handle.abort();
174    }
175
176    /// Await the exact classified terminal outcome.
177    pub async fn outcome(self) -> TaskOutcome<T> {
178        self.handle.outcome().await
179    }
180}
181
182impl<A, E, L, T> RootRetirement<ActorRef<A, MailboxSender<E>, L>, T>
183where
184    E: ShutdownEvent,
185    L: IncarnationReporter,
186{
187    /// Publish one typed graceful-shutdown request.
188    ///
189    /// # Errors
190    ///
191    /// Returns the typed protocol-construction or closed-mailbox failure.
192    pub fn request_shutdown(&self) -> Result<(), ShutdownRequestError> {
193        self.handle.actor_ref().request_shutdown()
194    }
195}
196
197/// Nameable retirement type produced for one behavior and system.
198#[doc(hidden)]
199pub type BehaviorRetirement<R, B, L = NoLifecycle> =
200    RootRetirement<BehaviorRef<B, BehaviorReporter<R, B, L>>, RootOutcome<R, B, L>>;
201
202/// Separate delivery and retirement seats returned by transactional activation.
203pub struct RootActivation<B: Behavior<Ph = Never>, R> {
204    /// Cloneable delivery-only capability.
205    pub endpoint: RootEndpoint<B>,
206    /// Affine graceful/forced retirement authority.
207    pub retirement: R,
208}
209
210/// Nameable transactional activation result for one behavior and system.
211#[doc(hidden)]
212pub type BehaviorActivation<R, B, L = NoLifecycle> = RootActivation<B, BehaviorRetirement<R, B, L>>;
213
214/// Result of constructing and spawning one behavior actor.
215#[doc(hidden)]
216pub type BehaviorSpawnResult<R, B, L = NoLifecycle> = Result<
217    Handle<
218        BehaviorRef<B, BehaviorReporter<R, B, L>>,
219        ActorResult<B, BehaviorEnvironment<R, B, NoParent, L>>,
220    >,
221    <R as EndpointRegistry<AddrOf<B>, <B as Behavior>::Msg, AnchorEndpoint<B>>>::Error,
222>;
223
224/// Prepared-but-unlaunched ownership state for one behavior actor.
225#[doc(hidden)]
226type BehaviorPreparation<R, B, P, L> = PreparedIncarnation<
227    B,
228    BehaviorEnvironment<R, B, P, L>,
229    BehaviorRegistration<R, B>,
230    BehaviorRef<B, BehaviorReporter<R, B, L>>,
231    ActorResult<B, BehaviorEnvironment<R, B, P, L>>,
232    AddrOf<B>,
233    BehaviorReporter<R, B, L>,
234>;
235
236/// Provisional, unregistered ownership state for one behavior actor.
237#[doc(hidden)]
238type BehaviorProvisional<R, B, P, L> = ProvisionalIncarnation<
239    B,
240    BehaviorEnvironment<R, B, P, L>,
241    ActorResult<B, BehaviorEnvironment<R, B, P, L>>,
242    AddrOf<B>,
243    AnchorOf<B>,
244    MailboxSender<EventOf<B>>,
245>;
246
247/// Failure of one transactional child birth.
248///
249/// The exact stage is preserved: registration collisions, initialization
250/// fold failures, and initialization effect failures classify differently
251/// for same-action rejection delivery.
252#[doc(hidden)]
253#[derive(Debug, thiserror::Error)]
254pub enum SystemBirthError<R, I, F> {
255    /// The address generation could not be claimed at commit.
256    #[error("child address registration failed: {0:?}")]
257    Registration(R),
258    /// The synchronous initialization fold failed.
259    #[error("child initialization failed: {0:?}")]
260    Initialization(I),
261    /// An initialization effect failed during interpretation.
262    #[error("child initialization effect failed: {0:?}")]
263    Effects(F),
264}
265
266impl<R, I, F> From<RunError<I, F>> for SystemBirthError<R, I, F> {
267    fn from(error: RunError<I, F>) -> Self {
268        match error {
269            RunError::Behavior(initialization) => Self::Initialization(initialization),
270            RunError::Environment(effects) => Self::Effects(effects),
271            RunError::Poisoned => unreachable!("init cannot poison the executor"),
272        }
273    }
274}
275
276impl<R, I, F> CreationFailure for SystemBirthError<R, I, F> {
277    fn rejection(&self) -> behavior::CreationRejection {
278        match self {
279            Self::Registration(_) | Self::Effects(_) => {
280                behavior::CreationRejection::EnvironmentFailed
281            }
282            Self::Initialization(_) => behavior::CreationRejection::InitializationFailed,
283        }
284    }
285}
286
287impl<R: Clone, L: Clone> System<R, L> {
288    /// Initialize, register, and launch one exact root incarnation transactionally.
289    ///
290    /// # Errors
291    ///
292    /// Preserves initialization-fold, initialization-effect, and registration
293    /// failures as separate variants. No endpoint is registered when either
294    /// initialization stage fails.
295    pub async fn activate<B>(
296        &self,
297        address: AddrOf<B>,
298        behavior: B,
299    ) -> Result<
300        BehaviorActivation<R, B, L>,
301        SystemBirthError<
302            <R as EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>>>::Error,
303            B::Error,
304            <BehaviorEnvironment<R, B, NoParent, L> as Environment>::Error,
305        >,
306    >
307    where
308        B: Behavior<Ph = Never> + Send + 'static,
309        AddrOf<B>: Send + Sync + 'static,
310        NonceOf<B>: Send + 'static,
311        B::Sends: RouteSends<AddrOf<B>, EffectsOf<R, B, Self, NoParent>>
312            + ObservesCreations<NonceOf<B>>
313            + Send
314            + 'static,
315        <B::Sends as RouteSends<AddrOf<B>, EffectsOf<R, B, Self, NoParent>>>::Error:
316            Send + Sync + 'static,
317        B::Birth: RuntimeBirthMode<AddrOf<B>, Self, AnchorOf<B>> + 'static,
318        ChildOf<B>: Send + 'static,
319        RuntimeOf<B, Self>: Send + Sync + 'static,
320        <RuntimeOf<B, Self> as ChildRuntime<AddrOf<B>, ChildOf<B>, AnchorOf<B>>>::Error:
321            CreationFailure + Send + Sync + 'static,
322        B::Event: ShutdownEvent + TimeEvent + Send + 'static,
323        B::Msg: Send + 'static,
324        B::Error: Send + Sync + 'static,
325        R: Clone + EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>> + Send + Sync + 'static,
326        BehaviorRegistration<R, B>: Send + 'static,
327        L: LifecycleFactory<AddrOf<B>, BehaviorRegistration<R, B>>,
328    {
329        let mut provisional = self.prepare_provisional(address, behavior, NoParent);
330        let pending_exit = match provisional.driver().run_init().await {
331            Ok(exit) => exit,
332            Err(error) => {
333                provisional.driver().retire().await;
334                return Err(SystemBirthError::from(error));
335            }
336        };
337        let prepared = provisional
338            .commit(&self.router, &self.lifecycle)
339            .map_err(SystemBirthError::Registration)?;
340        let actor_ref = prepared.actor_ref().clone();
341        let endpoint = RootEndpoint {
342            inner: ActorRef::new(address, actor_ref.sender_anchor()),
343        };
344        let handle = prepared.launch(LaunchMode::Initialized(pending_exit), false);
345        Ok(RootActivation {
346            endpoint,
347            retirement: RootRetirement { handle },
348        })
349    }
350
351    /// Construct and spawn one behavior actor.
352    ///
353    /// # Errors
354    ///
355    /// Returns endpoint-registration failure, including address collision.
356    ///
357    /// # Panics
358    ///
359    /// Panics if called outside a Tokio runtime, or if a freshly allocated
360    /// private Bombay Observe namespace cannot resolve the subject registered
361    /// immediately beforehand.
362    pub fn spawn<B>(&self, address: AddrOf<B>, behavior: B) -> BehaviorSpawnResult<R, B, L>
363    where
364        B: Behavior<Ph = Never> + Send + 'static,
365        AddrOf<B>: Send + Sync + 'static,
366        NonceOf<B>: Send + 'static,
367        B::Sends: RouteSends<AddrOf<B>, EffectsOf<R, B, Self, NoParent>>
368            + ObservesCreations<NonceOf<B>>
369            + Send
370            + 'static,
371        <B::Sends as RouteSends<AddrOf<B>, EffectsOf<R, B, Self, NoParent>>>::Error:
372            Send + Sync + 'static,
373        B::Birth: RuntimeBirthMode<AddrOf<B>, Self, AnchorOf<B>> + 'static,
374        ChildOf<B>: Send + 'static,
375        RuntimeOf<B, Self>: Send + Sync + 'static,
376        <RuntimeOf<B, Self> as ChildRuntime<AddrOf<B>, ChildOf<B>, AnchorOf<B>>>::Error:
377            CreationFailure + Send + Sync + 'static,
378        B::Event: TimeEvent + Send + 'static,
379        B::Msg: Send + 'static,
380        B::Error: Send + Sync + 'static,
381        R: Clone + EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>> + Send + Sync + 'static,
382        BehaviorRegistration<R, B>: Send + 'static,
383        L: LifecycleFactory<AddrOf<B>, BehaviorRegistration<R, B>>,
384    {
385        let prepared = self.prepare(address, behavior, NoParent)?;
386        Ok(prepared.launch(LaunchMode::Uninitialized, false))
387    }
388
389    #[allow(
390        clippy::type_complexity,
391        reason = "the private result retains the exact typed registration error"
392    )]
393    fn prepare<B, Parent>(
394        &self,
395        address: AddrOf<B>,
396        behavior: B,
397        parent: Parent,
398    ) -> Result<
399        BehaviorPreparation<R, B, Parent, L>,
400        <R as EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>>>::Error,
401    >
402    where
403        B: Behavior<Ph = Never> + Send + 'static,
404        AddrOf<B>: Send + Sync + 'static,
405        NonceOf<B>: Send + 'static,
406        B::Sends: RouteSends<AddrOf<B>, EffectsOf<R, B, Self, Parent>>
407            + ObservesCreations<NonceOf<B>>
408            + Send
409            + 'static,
410        <B::Sends as RouteSends<AddrOf<B>, EffectsOf<R, B, Self, Parent>>>::Error:
411            Send + Sync + 'static,
412        B::Birth: RuntimeBirthMode<AddrOf<B>, Self, AnchorOf<B>> + 'static,
413        ChildOf<B>: Send + 'static,
414        RuntimeOf<B, Self>: Send + Sync + 'static,
415        <RuntimeOf<B, Self> as ChildRuntime<AddrOf<B>, ChildOf<B>, AnchorOf<B>>>::Error:
416            CreationFailure + Send + Sync + 'static,
417        B::Event: TimeEvent + Send + 'static,
418        B::Msg: Send + 'static,
419        B::Error: Send + Sync + 'static,
420        Parent: Send + 'static,
421        R: Clone + EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>> + Send + Sync + 'static,
422        BehaviorRegistration<R, B>: Send + 'static,
423        L: LifecycleFactory<AddrOf<B>, BehaviorRegistration<R, B>>,
424    {
425        self.prepare_provisional(address, behavior, parent)
426            .commit(&self.router, &self.lifecycle)
427    }
428
429    /// Prepare every actor resource without claiming the address generation.
430    fn prepare_provisional<B, Parent>(
431        &self,
432        address: AddrOf<B>,
433        behavior: B,
434        parent: Parent,
435    ) -> BehaviorProvisional<R, B, Parent, L>
436    where
437        B: Behavior<Ph = Never> + Send + 'static,
438        AddrOf<B>: Send + Sync + 'static,
439        NonceOf<B>: Send + 'static,
440        B::Sends: RouteSends<AddrOf<B>, EffectsOf<R, B, Self, Parent>>
441            + ObservesCreations<NonceOf<B>>
442            + Send
443            + 'static,
444        <B::Sends as RouteSends<AddrOf<B>, EffectsOf<R, B, Self, Parent>>>::Error:
445            Send + Sync + 'static,
446        B::Birth: RuntimeBirthMode<AddrOf<B>, Self, AnchorOf<B>> + 'static,
447        ChildOf<B>: Send + 'static,
448        RuntimeOf<B, Self>: Send + Sync + 'static,
449        <RuntimeOf<B, Self> as ChildRuntime<AddrOf<B>, ChildOf<B>, AnchorOf<B>>>::Error:
450            CreationFailure + Send + Sync + 'static,
451        B::Event: TimeEvent + Send + 'static,
452        B::Msg: Send + 'static,
453        B::Error: Send + Sync + 'static,
454        Parent: Send + 'static,
455        R: Clone + EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>> + Send + Sync + 'static,
456        BehaviorRegistration<R, B>: Send + 'static,
457        L: LifecycleFactory<AddrOf<B>, BehaviorRegistration<R, B>>,
458    {
459        let observation_space = ObservationSpace::new();
460        let subject = observation_space
461            .subject(())
462            .expect("a fresh completion namespace must be vacant");
463        let observation = observation_space
464            .observe(&())
465            .expect("a newly registered completion subject must resolve");
466        let peer_space = ObservationSpace::new();
467        let peer_subject = peer_space
468            .subject(())
469            .expect("a fresh peer-completion namespace must be vacant");
470        let (sender, source) = self.mailbox.create::<B::Event>();
471        let cancellation_requested = Arc::new(AtomicBool::new(false));
472        let response = sender.anchor();
473        let endpoint =
474            IncarnationEndpoint::new(ActorRef::new(address, response.clone()), peer_space);
475        let children =
476            <B::Birth as RuntimeBirthMode<AddrOf<B>, Self, AnchorOf<B>>>::runtime(self.clone());
477        let environment = ActorEnvironment::new(
478            address,
479            source,
480            self.router.clone(),
481            children,
482            response,
483            parent,
484        );
485        ProvisionalIncarnation::new(
486            Driver::new(behavior, environment),
487            address,
488            endpoint,
489            sender,
490            subject,
491            observation,
492            peer_subject,
493            cancellation_requested,
494        )
495    }
496}
497
498impl<R, L, B, ParentSink> ChildRuntime<AddrOf<B>, B, ParentSink> for SystemChildren<System<R, L>>
499where
500    B: Behavior<Ph = Never> + Send + 'static,
501    AddrOf<B>: Send + Sync + 'static,
502    NonceOf<B>: Send + 'static,
503    B::Sends: RouteSends<AddrOf<B>, EffectsOf<R, B, System<R, L>, ParentReporter<AddrOf<B>, ParentSink>>>
504        + ObservesCreations<NonceOf<B>>
505        + Send
506        + 'static,
507    <B::Sends as RouteSends<
508        AddrOf<B>,
509        EffectsOf<R, B, System<R, L>, ParentReporter<AddrOf<B>, ParentSink>>,
510    >>::Error: Send + Sync + 'static,
511    B::Birth: RuntimeBirthMode<AddrOf<B>, System<R, L>, AnchorOf<B>> + 'static,
512    ChildOf<B>: Send + 'static,
513    RuntimeOf<B, System<R, L>>: Send + Sync + 'static,
514    <RuntimeOf<B, System<R, L>> as ChildRuntime<AddrOf<B>, ChildOf<B>, AnchorOf<B>>>::Error:
515        CreationFailure + Send + Sync + 'static,
516    B::Event: ShutdownEvent + TimeEvent + Send + 'static,
517    B::Msg: Send + 'static,
518    B::Error: Send + Sync + 'static,
519    R: Clone + EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>> + Send + Sync + 'static,
520    BehaviorRegistration<R, B>: Send + 'static,
521    L: Clone + LifecycleFactory<AddrOf<B>, BehaviorRegistration<R, B>> + Send + Sync + 'static,
522    ParentSink: EventSender + Clone + Send + Sync + 'static,
523    ParentSink::Event: Send + 'static,
524{
525    type Lease = ChildLease<
526        BehaviorRef<B, BehaviorReporter<R, B, L>>,
527        ActorResult<B, BehaviorEnvironment<R, B, ParentReporter<AddrOf<B>, ParentSink>, L>>,
528    >;
529    type Error = SystemBirthError<
530        <R as EndpointRegistry<AddrOf<B>, B::Msg, AnchorEndpoint<B>>>::Error,
531        B::Error,
532        <BehaviorEnvironment<R, B, ParentReporter<AddrOf<B>, ParentSink>, L> as Environment>::Error,
533    >;
534
535    async fn birth(
536        &self,
537        parent: AddrOf<B>,
538        child: behavior::Create<AddrOf<B>, B>,
539        response: ParentSink,
540    ) -> Result<Self::Lease, Self::Error> {
541        let address = parent.birth(child.nonce);
542        let kind = child.kind;
543        let reporter = ParentReporter::new(child.nonce, response);
544        let mut provisional = self
545            .system
546            .prepare_provisional(address, child.child, reporter);
547        let pending_exit = match provisional.driver().run_init().await {
548            Ok(exit) => exit,
549            Err(error) => {
550                provisional.driver().retire().await;
551                return Err(SystemBirthError::from(error));
552            }
553        };
554        let prepared = provisional
555            .commit(&self.system.router, &self.system.lifecycle)
556            .map_err(SystemBirthError::Registration)?;
557        let restarted = matches!(kind, behavior::CreationKind::ReplacementIncarnation { .. });
558        let handle = prepared.launch(LaunchMode::Initialized(pending_exit), restarted);
559        Ok(handle.into_child_lease())
560    }
561}
562
563#[cfg(test)]
564mod tests {
565    use std::convert::Infallible;
566
567    use crate::{
568        ActorRef, AddressRouter, DeliveryRouter, EndpointRegistry, IncarnationEndpoint,
569        MailboxConfig, RunExit, System, TaskOutcome,
570    };
571    use behavior::{
572        Actions, Address, Behavior, Births, Create, Delivery, Handler, MailAddr, Never, NoBirths,
573        Pure, Recipient, ServiceSends, UnwatchPeer, User, WatchEvent,
574    };
575
576    struct StopWithMessage;
577
578    struct CancelUnknownPeer;
579
580    impl Behavior for CancelUnknownPeer {
581        type Addr = MailAddr;
582        type Msg = u8;
583        type Event = WatchEvent<User<MailAddr, u8>>;
584        type Sends = ServiceSends<UnwatchPeer<MailAddr>>;
585        type Ph = Never;
586        type Error = Never;
587        type Birth = NoBirths;
588
589        fn init(&mut self) -> behavior::BehaviorActed<Self> {
590            Ok(Actions::new(
591                ServiceSends::one(UnwatchPeer::new(MailAddr(8))),
592                Vec::new(),
593                behavior::Step::Stop(behavior::Exit::Normal),
594            ))
595        }
596
597        fn transition(&mut self, _event: Self::Event) -> behavior::BehaviorActed<Self> {
598            unreachable!("the initialization fold stops")
599        }
600    }
601
602    struct ForwardTo(MailAddr);
603
604    impl Handler for StopWithMessage {
605        type Addr = MailAddr;
606        type Msg = u8;
607
608        fn receive(
609            &mut self,
610            _from: MailAddr,
611            _message: u8,
612        ) -> behavior::Acted<MailAddr, Never, Vec<Delivery<MailAddr, Never>>, NoBirths, Never>
613        {
614            Ok(Actions::stop(behavior::Exit::Normal))
615        }
616    }
617
618    impl Handler<u8> for ForwardTo {
619        type Addr = MailAddr;
620        type Msg = u8;
621
622        fn receive(
623            &mut self,
624            _from: MailAddr,
625            message: u8,
626        ) -> behavior::Acted<MailAddr, Never, Vec<Delivery<MailAddr, u8>>, NoBirths, Never>
627        {
628            Ok(Actions {
629                sends: vec![Delivery::new(Recipient::global(self.0), message + 1)],
630                creates: Vec::new(),
631                become_: behavior::Step::Stop(behavior::Exit::Normal),
632            })
633        }
634    }
635
636    struct StopAfterReceiving;
637
638    struct BirthAndSend {
639        receiver: MailAddr,
640    }
641
642    struct ForwardChild {
643        receiver: MailAddr,
644    }
645
646    impl Handler<u8, Births<Pure<ForwardChild, u8>>> for BirthAndSend {
647        type Addr = MailAddr;
648        type Msg = u8;
649
650        fn receive(
651            &mut self,
652            _from: MailAddr,
653            message: u8,
654        ) -> behavior::Acted<
655            MailAddr,
656            Never,
657            Vec<Delivery<MailAddr, u8>>,
658            Births<Pure<ForwardChild, u8>>,
659            Never,
660        > {
661            Ok(Actions {
662                sends: vec![Delivery::new(
663                    Recipient::global(MailAddr(1).birth(7)),
664                    message,
665                )],
666                creates: vec![Create::birth(
667                    7,
668                    Pure::new(ForwardChild {
669                        receiver: self.receiver,
670                    }),
671                )],
672                become_: behavior::Step::Stop(behavior::Exit::Normal),
673            })
674        }
675    }
676
677    impl Handler<u8> for ForwardChild {
678        type Addr = MailAddr;
679        type Msg = u8;
680
681        fn receive(
682            &mut self,
683            _from: MailAddr,
684            message: u8,
685        ) -> behavior::Acted<MailAddr, Never, Vec<Delivery<MailAddr, u8>>, NoBirths, Never>
686        {
687            Ok(Actions {
688                sends: vec![Delivery::new(Recipient::global(self.receiver), message + 1)],
689                creates: Vec::new(),
690                become_: behavior::Step::Stop(behavior::Exit::Normal),
691            })
692        }
693    }
694
695    impl Handler<u8> for StopAfterReceiving {
696        type Addr = MailAddr;
697        type Msg = u8;
698
699        fn receive(
700            &mut self,
701            _from: MailAddr,
702            _message: u8,
703        ) -> behavior::Acted<MailAddr, Never, Vec<Delivery<MailAddr, u8>>, NoBirths, Never>
704        {
705            Ok(Actions::stop(behavior::Exit::Normal))
706        }
707    }
708
709    #[derive(Clone, Copy)]
710    struct NoRouter;
711
712    impl DeliveryRouter<MailAddr, Never> for NoRouter {
713        type Error = Infallible;
714
715        async fn deliver(
716            &self,
717            _from: MailAddr,
718            delivery: Delivery<MailAddr, Never>,
719        ) -> Result<(), Self::Error> {
720            match delivery.message {}
721        }
722    }
723
724    impl<S> EndpointRegistry<MailAddr, u8, IncarnationEndpoint<MailAddr, ActorRef<MailAddr, S>>>
725        for NoRouter
726    {
727        type Error = Infallible;
728        type Registration = ();
729
730        fn register(
731            &self,
732            _address: MailAddr,
733            _endpoint: IncarnationEndpoint<MailAddr, ActorRef<MailAddr, S>>,
734        ) -> Result<Self::Registration, Self::Error> {
735            Ok(())
736        }
737    }
738
739    #[tokio::test]
740    async fn configured_system_constructs_mailbox_and_returns_actor_reference() {
741        let system = System::new(MailboxConfig::bounded(4), NoRouter);
742
743        let spawned = system
744            .spawn(MailAddr(9), Pure::new(StopWithMessage))
745            .unwrap();
746        spawned.actor_ref().send(MailAddr(7), 1).await.unwrap();
747
748        assert_eq!(
749            spawned.outcome().await,
750            TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
751        );
752    }
753
754    #[tokio::test]
755    async fn unwatch_service_composes_through_system_without_a_creation_lane() {
756        let system = System::new(MailboxConfig::bounded(1), NoRouter);
757        let spawned = system.spawn(MailAddr(9), CancelUnknownPeer).unwrap();
758
759        assert_eq!(
760            spawned.outcome().await,
761            TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
762        );
763    }
764
765    #[tokio::test]
766    async fn address_registration_lives_exactly_as_long_as_actor_task() {
767        let router = AddressRouter::default();
768        let system = System::new(MailboxConfig::bounded(4), router);
769
770        let first = system
771            .spawn(MailAddr(9), Pure::new(ForwardTo(MailAddr(88))))
772            .unwrap();
773        assert!(
774            system
775                .spawn(MailAddr(9), Pure::new(ForwardTo(MailAddr(88))))
776                .is_err(),
777            "a running actor owns its address generation"
778        );
779
780        first.actor_ref().send(MailAddr(7), 1).await.unwrap();
781        assert!(matches!(
782            first.outcome().await,
783            TaskOutcome::Returned(Err(_))
784        ));
785
786        let replacement = system
787            .spawn(MailAddr(9), Pure::new(ForwardTo(MailAddr(88))))
788            .expect("task exit releases the old address generation");
789        replacement.actor_ref().send(MailAddr(7), 1).await.unwrap();
790        assert!(matches!(
791            replacement.outcome().await,
792            TaskOutcome::Returned(Err(_))
793        ));
794    }
795
796    #[tokio::test]
797    async fn two_actors_exchange_a_typed_message_through_shared_routing() {
798        let router = AddressRouter::default();
799        let system = System::new(MailboxConfig::bounded(4), router);
800        let receiver = system
801            .spawn(MailAddr(2), Pure::new(StopAfterReceiving))
802            .unwrap();
803        let sender = system
804            .spawn(MailAddr(1), Pure::new(ForwardTo(MailAddr(2))))
805            .unwrap();
806
807        sender.actor_ref().send(MailAddr(0), 40).await.unwrap();
808
809        assert!(matches!(
810            sender.outcome().await,
811            TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
812        ));
813        assert!(matches!(
814            receiver.outcome().await,
815            TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
816        ));
817    }
818
819    #[tokio::test]
820    async fn mailbox_anchor_routes_without_pinning_mailbox_open() {
821        let router = AddressRouter::default();
822        let system = System::new(MailboxConfig::bounded(4), router);
823        let receiver = system
824            .spawn(MailAddr(2), Pure::new(StopAfterReceiving))
825            .unwrap();
826        let sender = system
827            .spawn(MailAddr(1), Pure::new(ForwardTo(MailAddr(2))))
828            .unwrap();
829
830        sender.actor_ref().send(MailAddr(0), 40).await.unwrap();
831        assert!(matches!(
832            sender.outcome().await,
833            TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
834        ));
835        assert!(matches!(
836            receiver.outcome().await,
837            TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
838        ));
839
840        let parked = system
841            .spawn(MailAddr(3), Pure::new(StopAfterReceiving))
842            .unwrap();
843        assert!(matches!(
844            parked.close().await,
845            TaskOutcome::Returned(Ok(RunExit::EnvironmentClosed))
846        ));
847        assert!(
848            system
849                .spawn(MailAddr(3), Pure::new(StopAfterReceiving))
850                .is_ok(),
851            "environment closure releases the address generation"
852        );
853    }
854
855    #[tokio::test]
856    async fn typed_creator_spawns_child_before_routing_same_transition_send() {
857        let router = AddressRouter::default();
858        let system = System::new(MailboxConfig::bounded(4), router);
859        let receiver = system
860            .spawn(MailAddr(9), Pure::new(StopAfterReceiving))
861            .unwrap();
862        let parent = system
863            .spawn(
864                MailAddr(1),
865                Pure::new(BirthAndSend {
866                    receiver: MailAddr(9),
867                }),
868            )
869            .unwrap();
870
871        parent.actor_ref().send(MailAddr(0), 40).await.unwrap();
872
873        assert!(matches!(
874            parent.outcome().await,
875            TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
876        ));
877        assert!(matches!(
878            receiver.outcome().await,
879            TaskOutcome::Returned(Ok(RunExit::Stopped(behavior::Exit::Normal)))
880        ));
881    }
882}