behavior/supervision/
proxy.rs

1//! Stable proxy lifecycle and fresh worker incarnation replacement.
2
3use super::protocol::{ProxyCommand, SupervisionEvent};
4use crate::behavior::{
5    Actions, Address, Behavior, Births, Create, Delivery, Recipient, SendAlgebra, SendProduct,
6    ServiceSends, User,
7};
8use crate::protocol::{ObserveChild, ReportWorkerStopped};
9use crate::verdict::{Never, Step};
10
11/// The stable actor. Every replacement is an ordinary fresh birth beneath it.
12///
13/// Each emitted worker birth is paired with an [`ObserveChild`] request. A
14/// matching [`ChildStopped`] leaves the proxy alive and emits a
15/// [`ReportWorkerStopped`] carrying the outcome unchanged. Stale child-stop
16/// observations are inert.
17pub struct Proxy<C: Behavior<Ph = Never>> {
18    worker: Option<C>,
19    generation: u64,
20    worker_alive: bool,
21    pending: Option<C>,
22}
23
24impl<C: Behavior<Ph = Never>> Proxy<C> {
25    #[must_use]
26    pub fn new(worker: C) -> Self {
27        Self {
28            worker: Some(worker),
29            generation: 0,
30            worker_alive: false,
31            pending: None,
32        }
33    }
34}
35
36impl<C> Behavior for Proxy<C>
37where
38    C: Behavior<Ph = Never> + Send,
39    C::Addr: Send,
40    <C::Addr as Address>::Nonce: From<u64> + Send,
41    C::Msg: Send,
42    C: Send,
43{
44    type Addr = C::Addr;
45    type Msg = ProxyCommand<C>;
46    type Event = SupervisionEvent<User<C::Addr, ProxyCommand<C>>, C::Addr>;
47    type Sends = SendProduct<
48        Vec<Delivery<C::Addr, C::Msg>>,
49        SendProduct<
50            ServiceSends<ObserveChild<C::Addr>>,
51            ServiceSends<ReportWorkerStopped<C::Addr>>,
52        >,
53    >;
54    type Ph = Never;
55    type Error = Never;
56    type Birth = Births<C>;
57
58    async fn init(&mut self) -> Result<Actions<C::Addr, Never, Self::Sends, Births<C>>, Never> {
59        let child = self.worker.take().expect("a proxy initializes once");
60        self.worker_alive = true;
61        Ok(Actions {
62            sends: SendProduct {
63                inner: Vec::new(),
64                own: SendProduct {
65                    inner: ServiceSends::one(ObserveChild {
66                        nonce: <C::Addr as Address>::Nonce::from(self.generation),
67                    }),
68                    own: ServiceSends::empty(),
69                },
70            },
71            creates: vec![Create::birth(
72                <C::Addr as Address>::Nonce::from(self.generation),
73                child,
74            )],
75            become_: Step::Continue,
76        })
77    }
78
79    async fn step(
80        &mut self,
81        event: Self::Event,
82    ) -> Result<Actions<C::Addr, Never, Self::Sends, Births<C>>, Never> {
83        let SupervisionEvent::Inner(event) = event else {
84            return match event {
85                SupervisionEvent::ChildStopped(event)
86                    if event.nonce == <C::Addr as Address>::Nonce::from(self.generation) =>
87                {
88                    self.worker_alive = false;
89                    let report = ReportWorkerStopped {
90                        outcome: event.outcome,
91                        at: event.at,
92                    };
93                    let creates = self.pending.take().map_or_else(Vec::new, |child| {
94                        self.generation = self
95                            .generation
96                            .checked_add(1)
97                            .expect("proxy generation exhausted");
98                        self.worker_alive = true;
99                        vec![Create::replacement_incarnation(
100                            <C::Addr as Address>::Nonce::from(self.generation),
101                            child,
102                        )]
103                    });
104                    let observes = creates
105                        .iter()
106                        .map(|create| ObserveChild {
107                            nonce: create.nonce,
108                        })
109                        .collect();
110                    Ok(Actions {
111                        sends: SendProduct {
112                            inner: Vec::new(),
113                            own: SendProduct {
114                                inner: ServiceSends::new(observes),
115                                own: ServiceSends::one(report),
116                            },
117                        },
118                        creates,
119                        become_: Step::Continue,
120                    })
121                }
122                SupervisionEvent::ChildStopped(_) | SupervisionEvent::WorkerStopped(_) => {
123                    Ok(Actions::cont())
124                }
125                SupervisionEvent::Inner(_) => unreachable!(),
126            };
127        };
128        match event.message {
129            ProxyCommand::Forward(message) => Ok(Actions {
130                sends: SendProduct {
131                    inner: self
132                        .worker_alive
133                        .then(|| {
134                            Delivery::new(
135                                Recipient::child(<C::Addr as Address>::Nonce::from(
136                                    self.generation,
137                                )),
138                                message,
139                            )
140                        })
141                        .into_iter()
142                        .collect(),
143                    own: SendProduct {
144                        inner: ServiceSends::empty(),
145                        own: ServiceSends::empty(),
146                    },
147                },
148                creates: Vec::new(),
149                become_: Step::Continue,
150            }),
151            ProxyCommand::Replace(child) => {
152                if self.worker_alive {
153                    self.pending = Some(child);
154                    return Ok(Actions::cont());
155                }
156                self.generation = self
157                    .generation
158                    .checked_add(1)
159                    .expect("proxy generation exhausted");
160                self.worker_alive = true;
161                let nonce = <C::Addr as Address>::Nonce::from(self.generation);
162                Ok(Actions {
163                    sends: SendProduct {
164                        inner: Vec::new(),
165                        own: SendProduct {
166                            inner: ServiceSends::one(ObserveChild { nonce }),
167                            own: ServiceSends::empty(),
168                        },
169                    },
170                    creates: vec![Create::replacement_incarnation(nonce, child)],
171                    become_: Step::Continue,
172                })
173            }
174        }
175    }
176}
behavior/supervision/proxy.rs

behavior/supervision/
proxy.rs
