behavior/
supervising.rs

1//! Pure supervision. Stable child identity is a proxy actor; replacement is a
2//! message to that proxy, which creates a fresh worker incarnation.
3
4use std::time::Duration;
5
6use tokio::time::Instant;
7
8use crate::behavior::{
9    Actions, Address, Behavior, Births, Create, Delivery, Recipient, SendAlgebra, SendProduct,
10    ServiceSends, User, UserEvent,
11};
12use crate::deadlined::{TimeEvent, TimeReached};
13use crate::verdict::{Never, Step};
14use crate::watching::{PeerEvent, PeerStopped};
15use crate::{Crash, Exit};
16
17#[derive(Debug, Clone, Copy, PartialEq, Eq)]
18pub enum Strategy {
19    OneForOne,
20    OneForAll,
21    RestForOne,
22}
23
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub enum RestartPolicy {
26    Permanent,
27    Transient,
28    Temporary,
29}
30
31#[must_use]
32pub const fn restart_one() -> Strategy {
33    Strategy::OneForOne
34}
35
36#[must_use]
37pub const fn restart_all() -> Strategy {
38    Strategy::OneForAll
39}
40
41#[must_use]
42pub const fn restart_rest() -> Strategy {
43    Strategy::RestForOne
44}
45
46#[derive(Debug, Clone, PartialEq, Eq)]
47pub struct ChildStopped<A: Address> {
48    pub nonce: A::Nonce,
49    pub outcome: Result<Exit<A>, Crash>,
50    pub at: Instant,
51}
52
53#[derive(Debug, Clone, Copy, PartialEq, Eq)]
54pub struct ObserveChild<A: Address> {
55    pub nonce: A::Nonce,
56}
57
58#[derive(Clone, PartialEq, Eq)]
59pub enum SupervisionEvent<E, A: Address> {
60    Inner(E),
61    ChildStopped(ChildStopped<A>),
62}
63
64pub trait ChildEvent<A: Address>: Sized {
65    fn child_stopped(event: ChildStopped<A>) -> Option<Self>;
66}
67
68impl<E, A: Address> ChildEvent<A> for SupervisionEvent<E, A> {
69    fn child_stopped(event: ChildStopped<A>) -> Option<Self> {
70        Some(Self::ChildStopped(event))
71    }
72}
73
74impl<E: UserEvent, A: Address> UserEvent for SupervisionEvent<E, A> {
75    type Addr = E::Addr;
76    type Message = E::Message;
77
78    fn user(from: Self::Addr, message: Self::Message) -> Self {
79        Self::Inner(E::user(from, message))
80    }
81
82    fn into_user(self) -> Result<User<Self::Addr, Self::Message>, Self> {
83        match self {
84            Self::Inner(event) => event.into_user().map_err(Self::Inner),
85            stopped @ Self::ChildStopped(_) => Err(stopped),
86        }
87    }
88}
89
90impl<E: TimeEvent, A: Address> TimeEvent for SupervisionEvent<E, A> {
91    fn time_reached(event: TimeReached) -> Option<Self> {
92        E::time_reached(event).map(Self::Inner)
93    }
94}
95
96impl<E: PeerEvent<A>, A: Address> PeerEvent<A> for SupervisionEvent<E, A> {
97    fn peer_stopped(event: PeerStopped<A>) -> Option<Self> {
98        E::peer_stopped(event).map(Self::Inner)
99    }
100}
101
102#[derive(Debug)]
103pub enum ProxyCommand<C: Behavior> {
104    Forward(C::Msg),
105    Replace(C),
106}
107
108pub type SupervisorSends<A, Sends, C> = SendProduct<
109    Sends,
110    SendProduct<ServiceSends<ObserveChild<A>>, Vec<Delivery<A, ProxyCommand<C>>>>,
111>;
112
113pub type SupervisorActions<B, C> = Actions<
114    <B as Behavior>::Addr,
115    <B as Behavior>::Ph,
116    SupervisorSends<<B as Behavior>::Addr, <B as Behavior>::Sends, C>,
117    Births<Proxy<C>>,
118>;
119
120/// The stable actor. Every replacement is an ordinary fresh birth beneath it.
121pub struct Proxy<C: Behavior<Ph = Never>> {
122    worker: Option<C>,
123    generation: u64,
124}
125
126impl<C: Behavior<Ph = Never>> Proxy<C> {
127    #[must_use]
128    pub fn new(worker: C) -> Self {
129        Self {
130            worker: Some(worker),
131            generation: 0,
132        }
133    }
134}
135
136impl<C> Behavior for Proxy<C>
137where
138    C: Behavior<Ph = Never> + Send,
139    C::Addr: Send,
140    <C::Addr as Address>::Nonce: From<u64> + Send,
141    C::Msg: Send,
142    C: Send,
143{
144    type Addr = C::Addr;
145    type Msg = ProxyCommand<C>;
146    type Event = User<C::Addr, ProxyCommand<C>>;
147    type Sends = Vec<Delivery<C::Addr, C::Msg>>;
148    type Ph = Never;
149    type Error = Never;
150    type Birth = Births<C>;
151    type Effect = Actions<C::Addr, Never, Self::Sends, Births<C>>;
152    type Done = Exit<C::Addr>;
153
154    async fn init(&mut self) -> Result<Self::Effect, Never> {
155        let child = self.worker.take().expect("a proxy initializes once");
156        Ok(Actions {
157            sends: Vec::new(),
158            creates: vec![Create {
159                nonce: <C::Addr as Address>::Nonce::from(self.generation),
160                child,
161            }],
162            become_: Step::Continue,
163        })
164    }
165
166    async fn step(&mut self, event: Self::Event) -> Result<Self::Effect, Never> {
167        match event.message {
168            ProxyCommand::Forward(message) => Ok(Actions {
169                sends: vec![Delivery::new(
170                    Recipient::child(<C::Addr as Address>::Nonce::from(self.generation)),
171                    message,
172                )],
173                creates: Vec::new(),
174                become_: Step::Continue,
175            }),
176            ProxyCommand::Replace(child) => {
177                self.generation = self
178                    .generation
179                    .checked_add(1)
180                    .expect("proxy generation exhausted");
181                Ok(Actions {
182                    sends: Vec::new(),
183                    creates: vec![Create {
184                        nonce: <C::Addr as Address>::Nonce::from(self.generation),
185                        child,
186                    }],
187                    become_: Step::Continue,
188                })
189            }
190        }
191    }
192}
193
194struct Slot {
195    alive: bool,
196    sequence: u64,
197}
198
199pub struct Supervising<B: Behavior, C: Behavior<Ph = Never, Addr = B::Addr>> {
200    inner: B,
201    slots: Vec<(<B::Addr as Address>::Nonce, Slot)>,
202    configured_count: usize,
203    next_sequence: u64,
204    build: fn(usize) -> C,
205    strategy: Strategy,
206    policy: RestartPolicy,
207    max_restarts: u32,
208    window: Duration,
209    restarts: Vec<Instant>,
210}
211
212impl<B, C> Supervising<B, C>
213where
214    B: Behavior<Birth = Births<C>>,
215    C: Behavior<Ph = Never, Addr = B::Addr>,
216{
217    #[allow(clippy::too_many_arguments, reason = "hidden by Spec")]
218    /// Construct the concrete supervisor behavior hidden by `Spec`.
219    ///
220    /// # Panics
221    /// Panics only if a fleet index cannot be represented by `u64`.
222    #[must_use]
223    pub fn new(
224        inner: B,
225        nonces: fn(usize) -> <B::Addr as Address>::Nonce,
226        count: usize,
227        build: fn(usize) -> C,
228        strategy: Strategy,
229        policy: RestartPolicy,
230        max_restarts: u32,
231        window: Duration,
232    ) -> Self {
233        let slots = (0..count)
234            .map(|index| {
235                (
236                    nonces(index),
237                    Slot {
238                        alive: true,
239                        sequence: u64::try_from(index).expect("fleet index fits u64"),
240                    },
241                )
242            })
243            .collect();
244        Self {
245            inner,
246            slots,
247            configured_count: count,
248            next_sequence: u64::try_from(count).expect("fleet size fits u64"),
249            build,
250            strategy,
251            policy,
252            max_restarts,
253            window,
254            restarts: Vec::new(),
255        }
256    }
257
258    #[must_use]
259    pub fn with_strategy(mut self, strategy: Strategy) -> Self {
260        self.strategy = strategy;
261        self
262    }
263
264    #[must_use]
265    pub fn with_policy(mut self, policy: RestartPolicy) -> Self {
266        self.policy = policy;
267        self
268    }
269
270    #[must_use]
271    pub fn with_budget(mut self, max: u32, window: Duration) -> Self {
272        self.max_restarts = max;
273        self.window = window;
274        self
275    }
276
277    fn position(&self, nonce: <B::Addr as Address>::Nonce) -> Option<usize> {
278        self.slots.iter().position(|(known, _)| *known == nonce)
279    }
280
281    #[must_use]
282    /// Report whether a known supervised proxy is alive.
283    ///
284    /// # Panics
285    /// Panics when `nonce` is not part of this supervisor topology.
286    pub fn is_alive(&self, nonce: <B::Addr as Address>::Nonce) -> bool {
287        self.slots[self.position(nonce).expect("unknown supervised nonce")]
288            .1
289            .alive
290    }
291
292    #[must_use]
293    pub fn child_count(&self) -> usize {
294        self.slots.len()
295    }
296
297    #[must_use]
298    pub fn restarts_in_window(&self) -> usize {
299        self.restarts.len()
300    }
301
302    fn replacements(
303        &mut self,
304        event: &ChildStopped<B::Addr>,
305    ) -> Vec<Delivery<B::Addr, ProxyCommand<C>>> {
306        let dead = self
307            .position(event.nonce)
308            .expect("unknown supervised nonce");
309        let eligible = match self.policy {
310            RestartPolicy::Permanent => true,
311            RestartPolicy::Transient => {
312                !matches!(&event.outcome, Ok(Exit::Normal | Exit::Collected))
313            }
314            RestartPolicy::Temporary => false,
315        };
316        if !eligible {
317            self.slots[dead].1.alive = false;
318            return Vec::new();
319        }
320        if self.window != Duration::MAX {
321            self.restarts.retain(|stamp| {
322                event
323                    .at
324                    .checked_duration_since(*stamp)
325                    .is_none_or(|age| age <= self.window)
326            });
327        }
328        let sequence = self.slots[dead].1.sequence;
329        let candidates: Vec<usize> = match self.strategy {
330            Strategy::OneForOne => vec![dead],
331            Strategy::OneForAll => self
332                .slots
333                .iter()
334                .enumerate()
335                .filter_map(|(index, (_, slot))| slot.alive.then_some(index))
336                .collect(),
337            Strategy::RestForOne => self
338                .slots
339                .iter()
340                .enumerate()
341                .filter_map(|(index, (_, slot))| {
342                    (slot.alive && slot.sequence >= sequence).then_some(index)
343                })
344                .collect(),
345        };
346        if self.restarts.len() + candidates.len() > self.max_restarts as usize {
347            self.slots[dead].1.alive = false;
348            return Vec::new();
349        }
350        self.restarts
351            .resize(self.restarts.len() + candidates.len(), event.at);
352        candidates
353            .into_iter()
354            .map(|index| {
355                self.slots[index].1.alive = true;
356                Delivery::new(
357                    Recipient::child(self.slots[index].0),
358                    ProxyCommand::Replace((self.build)(index)),
359                )
360            })
361            .collect()
362    }
363
364    fn wrap(
365        &mut self,
366        actions: Actions<B::Addr, B::Ph, B::Sends, Births<C>>,
367    ) -> SupervisorActions<B, C> {
368        let born: Vec<_> = actions.creates.iter().map(|create| create.nonce).collect();
369        for create in &actions.creates {
370            assert!(
371                self.position(create.nonce).is_none(),
372                "a child birth nonce must be fresh"
373            );
374            self.slots.push((
375                create.nonce,
376                Slot {
377                    alive: true,
378                    sequence: self.next_sequence,
379                },
380            ));
381            self.next_sequence = self
382                .next_sequence
383                .checked_add(1)
384                .expect("birth sequence exhausted");
385        }
386        Actions {
387            sends: SendProduct {
388                inner: actions.sends,
389                own: SendProduct {
390                    inner: ServiceSends::new(
391                        born.into_iter()
392                            .map(|nonce| ObserveChild { nonce })
393                            .collect(),
394                    ),
395                    own: Vec::new(),
396                },
397            },
398            creates: actions
399                .creates
400                .into_iter()
401                .map(|create| Create {
402                    nonce: create.nonce,
403                    child: Proxy::new(create.child),
404                })
405                .collect(),
406            become_: actions.become_,
407        }
408    }
409}
410
411impl<B, C, A, Ph, Sends> Behavior for Supervising<B, C>
412where
413    A: Address + Send,
414    Sends: SendAlgebra,
415    B: Behavior<
416            Addr = A,
417            Ph = Ph,
418            Sends = Sends,
419            Birth = Births<C>,
420            Effect = Actions<A, Ph, Sends, Births<C>>,
421            Done = Exit<A>,
422        > + Send,
423    B::Event: ChildEvent<B::Addr> + Send,
424    A::Nonce: From<u64> + Send,
425    B::Msg: Send,
426    C: Behavior<Ph = Never, Addr = B::Addr> + Send,
427{
428    type Addr = A;
429    type Msg = B::Msg;
430    type Event = SupervisionEvent<B::Event, B::Addr>;
431    type Sends = SupervisorSends<A, Sends, C>;
432    type Ph = Ph;
433    type Error = B::Error;
434    type Birth = Births<Proxy<C>>;
435    type Effect = Actions<A, Ph, Self::Sends, Births<Proxy<C>>>;
436    type Done = Exit<A>;
437
438    async fn init(&mut self) -> Result<Self::Effect, B::Error> {
439        let actions = self.inner.init().await?;
440        let mut actions = self.wrap(actions);
441        actions
442            .creates
443            .extend(self.slots[..self.configured_count].iter().enumerate().map(
444                |(index, (nonce, _))| Create {
445                    nonce: *nonce,
446                    child: Proxy::new((self.build)(index)),
447                },
448            ));
449        actions.sends.own.inner.extend(
450            self.slots[..self.configured_count]
451                .iter()
452                .map(|(nonce, _)| ObserveChild { nonce: *nonce }),
453        );
454        Ok(actions)
455    }
456
457    async fn step(&mut self, event: Self::Event) -> Result<Self::Effect, B::Error> {
458        match event {
459            SupervisionEvent::ChildStopped(event) => Ok(Actions {
460                sends: SendProduct {
461                    inner: B::Sends::empty(),
462                    own: SendProduct {
463                        inner: ServiceSends::empty(),
464                        own: self.replacements(&event),
465                    },
466                },
467                creates: Vec::new(),
468                become_: Step::Continue,
469            }),
470            SupervisionEvent::Inner(event) => {
471                let actions = self.inner.step(event).await?;
472                Ok(self.wrap(actions))
473            }
474        }
475    }
476}
behavior/supervising.rs

behavior/
supervising.rs
