behavior/supervision/adapter/
proxy.rs

1//! Stable proxy lifecycle and fresh worker incarnation replacement.
2
3use super::super::domain::{Incarnation, IncarnationEffects, IncarnationPhase, IncarnationReport};
4use super::super::protocol::{ProxyCommand, ProxyEvent};
5use crate::behavior::{
6    Actions, Address, Behavior, Births, Create, Delivery, Recipient, SendAlgebra, ServiceSends,
7    User,
8};
9use crate::next::{Never, Step};
10use crate::protocol::{
11    ObserveChild, ObserveCreation, ReportWorkerCreationResolved, ReportWorkerStopped,
12};
13use crate::{Own, SendInput};
14
15/// The concrete, statically dispatched effect lanes emitted by a [`Proxy`].
16pub struct ProxySends<A: Address, M> {
17    pub deliveries: Vec<Delivery<A, M>>,
18    pub child_observations: ServiceSends<ObserveChild<A::Nonce>>,
19    pub creation_observations: ServiceSends<ObserveCreation<A::Nonce>>,
20    pub stopped_reports: ServiceSends<ReportWorkerStopped<A>>,
21    pub creation_reports: ServiceSends<ReportWorkerCreationResolved<A::Nonce>>,
22}
23
24pub type ProxyActions<C> = Actions<
25    <C as Behavior>::Addr,
26    Never,
27    ProxySends<<C as Behavior>::Addr, <C as Behavior>::Msg>,
28    Births<C>,
29>;
30
31impl<A: Address, M> SendAlgebra for ProxySends<A, M> {
32    fn empty() -> Self {
33        Self {
34            deliveries: Vec::new(),
35            child_observations: ServiceSends::empty(),
36            creation_observations: ServiceSends::empty(),
37            stopped_reports: ServiceSends::empty(),
38            creation_reports: ServiceSends::empty(),
39        }
40    }
41
42    fn append(&mut self, mut other: Self) {
43        self.deliveries.append(&mut other.deliveries);
44        self.child_observations.append(other.child_observations);
45        self.creation_observations
46            .append(other.creation_observations);
47        self.stopped_reports.append(other.stopped_reports);
48        self.creation_reports.append(other.creation_reports);
49    }
50}
51
52impl<A: Address, M> SendInput<Delivery<A, M>, Own> for ProxySends<A, M> {
53    fn emit(&mut self, input: Delivery<A, M>) {
54        self.deliveries.push(input);
55    }
56}
57
58impl<A: Address, M> SendInput<ObserveChild<A::Nonce>, Own> for ProxySends<A, M> {
59    fn emit(&mut self, input: ObserveChild<A::Nonce>) {
60        self.child_observations.send(input);
61    }
62}
63
64impl<A: Address, M> SendInput<ObserveCreation<A::Nonce>, Own> for ProxySends<A, M> {
65    fn emit(&mut self, input: ObserveCreation<A::Nonce>) {
66        self.creation_observations.send(input);
67    }
68}
69
70impl<A: Address, M> SendInput<ReportWorkerStopped<A>, Own> for ProxySends<A, M> {
71    fn emit(&mut self, input: ReportWorkerStopped<A>) {
72        self.stopped_reports.send(input);
73    }
74}
75
76impl<A: Address, M> SendInput<ReportWorkerCreationResolved<A::Nonce>, Own> for ProxySends<A, M> {
77    fn emit(&mut self, input: ReportWorkerCreationResolved<A::Nonce>) {
78        self.creation_reports.send(input);
79    }
80}
81
82/// A stable actor that serializes fresh worker-incarnation installation.
83///
84/// A worker is routable only in `Running`. Deadline most one creation can be
85/// `Installing`; stale or provenance-mismatched results are inert. Rejection
86/// leaves `last_installed` unchanged, so a later attempt still names the last
87/// incarnation that actually existed.
88pub struct Proxy<C: Behavior<Ph = Never>> {
89    incarnation: Incarnation<<C::Addr as Address>::Nonce, C>,
90}
91
92impl<C: Behavior<Ph = Never>> Proxy<C> {
93    #[must_use]
94    pub fn new(worker: C) -> Self {
95        Self {
96            incarnation: Incarnation::new(worker),
97        }
98    }
99
100    #[must_use]
101    pub const fn phase(&self) -> IncarnationPhase<<C::Addr as Address>::Nonce> {
102        self.incarnation.phase()
103    }
104}
105
106impl<C> Proxy<C>
107where
108    C: Behavior<Ph = Never>,
109    C::Addr: Address,
110    <C::Addr as Address>::Nonce: From<u64>,
111{
112    fn actions(
113        effects: IncarnationEffects<<C::Addr as Address>::Nonce, C, C::Msg>,
114        stopped: Option<&crate::ChildStopped<C::Addr>>,
115    ) -> ProxyActions<C> {
116        let mut sends = ProxySends::empty();
117        if let Some((incarnation, message)) = effects.delivery {
118            sends
119                .deliveries
120                .push(Delivery::new(Recipient::child(incarnation), message));
121        }
122        if let Some(report) = effects.report {
123            match report {
124                IncarnationReport::CreationResolved(resolved) => {
125                    sends.creation_reports.extend([resolved.into()]);
126                }
127                IncarnationReport::Stopped { incarnation } => {
128                    let event = stopped.expect("stop report originates from child stop input");
129                    sends.stopped_reports.extend([ReportWorkerStopped::from(
130                        crate::ChildStopped::new(incarnation, event.outcome, event.at),
131                    )]);
132                }
133            }
134        }
135        let creates = effects.creation.map_or_else(Vec::new, |creation| {
136            sends
137                .child_observations
138                .extend([ObserveChild::new(creation.attempt)]);
139            sends
140                .creation_observations
141                .extend([ObserveCreation::new(creation.attempt)]);
142            vec![Create::new(creation.attempt, creation.child, creation.kind)]
143        });
144        Actions::new(sends, creates, Step::Continue)
145    }
146}
147
148impl<C> Behavior for Proxy<C>
149where
150    C: Behavior<Ph = Never>,
151    <C::Addr as Address>::Nonce: From<u64>,
152{
153    type Addr = C::Addr;
154    type Msg = ProxyCommand<C>;
155    type Event = ProxyEvent<User<C::Addr, ProxyCommand<C>>>;
156    type Sends = ProxySends<C::Addr, C::Msg>;
157    type Ph = Never;
158    type Error = Never;
159    type Birth = Births<C>;
160
161    fn init(&mut self) -> Result<Actions<C::Addr, Never, Self::Sends, Births<C>>, Never> {
162        let effects = self
163            .incarnation
164            .initialize()
165            .expect("a proxy initializes once");
166        Ok(Self::actions(effects, None))
167    }
168
169    fn transition(
170        &mut self,
171        event: Self::Event,
172    ) -> Result<Actions<C::Addr, Never, Self::Sends, Births<C>>, Never> {
173        Ok(match event {
174            ProxyEvent::CreationResolved(resolved) => Self::actions(
175                self.incarnation
176                    .creation_resolved(resolved.nonce, resolved.kind, resolved.result),
177                None,
178            ),
179            ProxyEvent::ChildStopped(event) => {
180                let effects = self.incarnation.child_stopped(event.nonce);
181                Self::actions(effects, Some(&event))
182            }
183            ProxyEvent::Inner(event) => match event.message {
184                ProxyCommand::Forward(message) => {
185                    let effects = self.incarnation.forward(message);
186                    Self::actions(effects, None)
187                }
188                ProxyCommand::Replace(child) => {
189                    let effects = self.incarnation.replace(child);
190                    Self::actions(effects, None)
191                }
192            },
193        })
194    }
195}
behavior/supervision/adapter/proxy.rs

behavior/supervision/adapter/
proxy.rs
