behavior/supervision/domain/
incarnation.rs

1//! Pure lifecycle domain for one stable proxy's worker incarnation.
2
3use crate::{CreationKind, CreationRejection, CreationResolved};
4
5/// The complete lifecycle state of the worker behind one stable proxy.
6enum IncarnationState<N, C> {
7    Dormant {
8        initial: C,
9    },
10    Installing {
11        attempt: N,
12        kind: CreationKind<N>,
13    },
14    Running {
15        incarnation: N,
16        queued_replacement: Option<C>,
17    },
18    Vacant {
19        last_installed: Option<N>,
20    },
21}
22
23/// A copyable observation of the lifecycle without owned child specifications.
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub enum IncarnationPhase<N> {
26    Dormant,
27    Installing { attempt: N, kind: CreationKind<N> },
28    Running { incarnation: N },
29    AwaitingStop { incarnation: N },
30    Vacant { last_installed: Option<N> },
31}
32
33#[derive(Debug, Clone, Copy, PartialEq, Eq)]
34pub(crate) enum IncarnationError {
35    AlreadyInitialized,
36}
37
38/// A fresh child creation selected by the lifecycle transition.
39#[derive(Debug, Clone, PartialEq, Eq)]
40pub(crate) struct IncarnationCreation<N, C> {
41    pub attempt: N,
42    pub kind: CreationKind<N>,
43    pub child: C,
44}
45
46impl<N, C> IncarnationCreation<N, C> {
47    #[must_use]
48    pub const fn new(attempt: N, kind: CreationKind<N>, child: C) -> Self {
49        Self {
50            attempt,
51            kind,
52            child,
53        }
54    }
55}
56
57/// A lifecycle fact to report to the stable proxy's parent.
58#[derive(Debug, Clone, PartialEq, Eq)]
59pub(crate) enum IncarnationReport<N> {
60    CreationResolved(CreationResolved<N>),
61    Stopped { incarnation: N },
62}
63
64impl<N> IncarnationReport<N> {
65    #[must_use]
66    pub const fn creation_resolved(
67        incarnation: N,
68        kind: CreationKind<N>,
69        result: Result<(), CreationRejection>,
70    ) -> Self {
71        Self::CreationResolved(CreationResolved::new(incarnation, kind, result))
72    }
73
74    #[must_use]
75    pub const fn stopped(incarnation: N) -> Self {
76        Self::Stopped { incarnation }
77    }
78}
79
80/// Independent effects selected by one lifecycle transition.
81///
82/// `creation` and `report` are independent because accepting an exact stop
83/// can both report that stop and begin an already-queued replacement.
84#[derive(Debug, Clone, PartialEq, Eq)]
85pub(crate) struct IncarnationEffects<N, C, M> {
86    pub creation: Option<IncarnationCreation<N, C>>,
87    pub delivery: Option<(N, M)>,
88    pub report: Option<IncarnationReport<N>>,
89}
90
91impl<N, C, M> IncarnationEffects<N, C, M> {
92    #[must_use]
93    pub fn new(
94        creation: Option<IncarnationCreation<N, C>>,
95        delivery: Option<(N, M)>,
96        report: Option<IncarnationReport<N>>,
97    ) -> Self {
98        Self {
99            creation,
100            delivery,
101            report,
102        }
103    }
104
105    #[must_use]
106    pub fn none() -> Self {
107        Self {
108            creation: None,
109            delivery: None,
110            report: None,
111        }
112    }
113}
114
115/// The typed state machine for one proxy's sequence of fresh incarnations.
116pub(crate) struct Incarnation<N, C> {
117    state: IncarnationState<N, C>,
118    next_attempt: u64,
119}
120
121impl<N, C> Incarnation<N, C> {
122    #[must_use]
123    pub const fn new(initial: C) -> Self {
124        Self {
125            state: IncarnationState::Dormant { initial },
126            next_attempt: 0,
127        }
128    }
129}
130
131impl<N: Copy, C> Incarnation<N, C> {
132    #[must_use]
133    pub const fn phase(&self) -> IncarnationPhase<N> {
134        match &self.state {
135            IncarnationState::Dormant { .. } => IncarnationPhase::Dormant,
136            IncarnationState::Installing { attempt, kind } => IncarnationPhase::Installing {
137                attempt: *attempt,
138                kind: *kind,
139            },
140            IncarnationState::Running {
141                incarnation,
142                queued_replacement: Some(_),
143            } => IncarnationPhase::AwaitingStop {
144                incarnation: *incarnation,
145            },
146            IncarnationState::Running {
147                incarnation,
148                queued_replacement: None,
149            } => IncarnationPhase::Running {
150                incarnation: *incarnation,
151            },
152            IncarnationState::Vacant { last_installed } => IncarnationPhase::Vacant {
153                last_installed: *last_installed,
154            },
155        }
156    }
157}
158
159impl<N: Copy + From<u64> + PartialEq, C> Incarnation<N, C> {
160    /// Emit the initial fresh creation exactly once.
161    ///
162    /// # Errors
163    /// Returns [`IncarnationError::AlreadyInitialized`] after leaving
164    /// `Dormant`.
165    pub(crate) fn initialize<M>(
166        &mut self,
167    ) -> Result<IncarnationEffects<N, C, M>, IncarnationError> {
168        let IncarnationState::Dormant { .. } = self.state else {
169            return Err(IncarnationError::AlreadyInitialized);
170        };
171        let IncarnationState::Dormant { initial } = core::mem::replace(
172            &mut self.state,
173            IncarnationState::Vacant {
174                last_installed: None,
175            },
176        ) else {
177            unreachable!("state was matched as dormant")
178        };
179        Ok(self.begin(initial, CreationKind::Birth))
180    }
181
182    fn begin<M>(&mut self, child: C, kind: CreationKind<N>) -> IncarnationEffects<N, C, M> {
183        let attempt = N::from(self.next_attempt);
184        self.next_attempt = self
185            .next_attempt
186            .checked_add(1)
187            .expect("incarnation creation nonce exhausted");
188        self.state = IncarnationState::Installing { attempt, kind };
189        IncarnationEffects::new(
190            Some(IncarnationCreation::new(attempt, kind, child)),
191            None,
192            None,
193        )
194    }
195
196    pub(crate) fn creation_resolved<M>(
197        &mut self,
198        attempt: N,
199        kind: CreationKind<N>,
200        result: Result<(), CreationRejection>,
201    ) -> IncarnationEffects<N, C, M> {
202        let IncarnationState::Installing {
203            attempt: pending,
204            kind: pending_kind,
205        } = self.state
206        else {
207            return IncarnationEffects::none();
208        };
209        if attempt != pending || kind != pending_kind {
210            return IncarnationEffects::none();
211        }
212        self.state = match result {
213            Ok(()) => IncarnationState::Running {
214                incarnation: attempt,
215                queued_replacement: None,
216            },
217            Err(_) => IncarnationState::Vacant {
218                last_installed: match kind {
219                    CreationKind::Birth => None,
220                    CreationKind::ReplacementIncarnation { replaces } => Some(replaces),
221                },
222            },
223        };
224        IncarnationEffects::new(
225            None,
226            None,
227            Some(IncarnationReport::creation_resolved(attempt, kind, result)),
228        )
229    }
230
231    pub(crate) fn child_stopped<M>(&mut self, stopped: N) -> IncarnationEffects<N, C, M> {
232        let IncarnationState::Running { incarnation, .. } = self.state else {
233            return IncarnationEffects::none();
234        };
235        if stopped != incarnation {
236            return IncarnationEffects::none();
237        }
238        let IncarnationState::Running {
239            incarnation,
240            queued_replacement,
241        } = core::mem::replace(
242            &mut self.state,
243            IncarnationState::Vacant {
244                last_installed: Some(incarnation),
245            },
246        )
247        else {
248            unreachable!("state was matched as running")
249        };
250        let mut effects = match queued_replacement {
251            Some(child) => self.begin(
252                child,
253                CreationKind::ReplacementIncarnation {
254                    replaces: incarnation,
255                },
256            ),
257            None => IncarnationEffects::none(),
258        };
259        effects.report = Some(IncarnationReport::stopped(incarnation));
260        effects
261    }
262
263    pub(crate) fn forward<M>(&self, message: M) -> IncarnationEffects<N, C, M> {
264        let delivery = match self.state {
265            IncarnationState::Running { incarnation, .. } => Some((incarnation, message)),
266            IncarnationState::Dormant { .. }
267            | IncarnationState::Installing { .. }
268            | IncarnationState::Vacant { .. } => None,
269        };
270        IncarnationEffects::new(None, delivery, None)
271    }
272
273    pub(crate) fn replace<M>(&mut self, child: C) -> IncarnationEffects<N, C, M> {
274        match &mut self.state {
275            IncarnationState::Running {
276                queued_replacement: queued_replacement @ None,
277                ..
278            } => {
279                *queued_replacement = Some(child);
280                IncarnationEffects::none()
281            }
282            IncarnationState::Vacant {
283                last_installed: Some(last),
284            } => {
285                let replaces = *last;
286                self.begin(child, CreationKind::ReplacementIncarnation { replaces })
287            }
288            IncarnationState::Dormant { .. }
289            | IncarnationState::Installing { .. }
290            | IncarnationState::Running { .. }
291            | IncarnationState::Vacant {
292                last_installed: None,
293            } => IncarnationEffects::none(),
294        }
295    }
296}
297
298#[cfg(test)]
299mod tests {
300    use super::*;
301
302    #[test]
303    fn rejected_attempt_preserves_successful_provenance() {
304        let mut machine = Incarnation::<u64, &'static str>::new("first");
305        machine.initialize::<()>().unwrap();
306        machine.creation_resolved::<()>(0, CreationKind::Birth, Ok(()));
307        machine.replace::<()>("second");
308        machine.child_stopped::<()>(0);
309        machine.creation_resolved::<()>(
310            1,
311            CreationKind::ReplacementIncarnation { replaces: 0 },
312            Err(CreationRejection::EnvironmentFailed),
313        );
314
315        assert_eq!(
316            machine.phase(),
317            IncarnationPhase::Vacant {
318                last_installed: Some(0)
319            }
320        );
321        let effects = machine.replace::<()>("third");
322        let creation = effects.creation.expect("replacement begins");
323        assert_eq!(creation.attempt, 2);
324        assert_eq!(
325            creation.kind,
326            CreationKind::ReplacementIncarnation { replaces: 0 }
327        );
328    }
329
330    #[test]
331    fn stale_inputs_are_inert() {
332        let mut machine = Incarnation::<u64, ()>::new(());
333        machine.initialize::<()>().unwrap();
334        let effects = machine.creation_resolved::<()>(9, CreationKind::Birth, Ok(()));
335        assert!(effects.creation.is_none());
336        assert!(effects.delivery.is_none());
337        assert!(effects.report.is_none());
338        assert_eq!(
339            machine.phase(),
340            IncarnationPhase::Installing {
341                attempt: 0,
342                kind: CreationKind::Birth
343            }
344        );
345    }
346}
behavior/supervision/domain/incarnation.rs

behavior/supervision/domain/
incarnation.rs
