made_core/entities/ceremony_instance/fold/
mod.rs

1//! Folding: the events of a session, back into its state.
2//!
3//! [`CeremonyInstance::apply`] writes what one event says and nothing
4//! else: no definition, no clock, no validation. Every rule was held
5//! when the event was decided, and re-checking here would make a
6//! stream that once folded stop folding the day a rule changed. One
7//! file per event family; [`CeremonyInstance::rehydrate`] is the
8//! whole fold from the opening event on.
9
10use crate::entities::{CeremonyEvent, CeremonyInstance};
11use crate::error::DomainError;
12
13mod children;
14mod context;
15mod execution_receipts;
16mod guard_decisions;
17mod import;
18mod interventions;
19mod lifecycle;
20mod participant_bindings;
21mod reasons;
22mod recollection;
23mod start;
24mod step_execution;
25mod succession;
26mod transitions;
27
28impl CeremonyInstance {
29    /// Write one event into this session.
30    ///
31    /// Infallible and total: an event the session cannot honour —
32    /// a response to an item it does not hold, a second opening —
33    /// leaves it untouched rather than failing, because a fold is
34    /// not where a bad stream gets refused. `updated_at` becomes the
35    /// event's own timestamp.
36    pub fn apply(&mut self, event: &CeremonyEvent) {
37        match event {
38            CeremonyEvent::HostHandoffRecorded(recorded) => {
39                self.host_handoffs
40                    .insert(recorded.declaration.id.clone(), recorded.clone());
41                self.updated_at = recorded.recorded_at;
42            }
43            // A stream opens once. Applying the opening to a session
44            // that already exists is a programming error in the
45            // caller, and the session is left exactly as it was: a
46            // fold that panicked over it would take a host down over
47            // one bad stream, and one that reopened the session would
48            // silently discard everything after the first opening.
49            CeremonyEvent::CeremonyInstanceStarted(_) => {}
50            CeremonyEvent::ParticipantsBound(bound) => self.apply_participants_bound(bound),
51            CeremonyEvent::StepStarted(started) => self.apply_step_started(started),
52            CeremonyEvent::StepLeaseRenewed(renewed) => {
53                if let Some(request) = &renewed.request {
54                    self.lease_renewals
55                        .insert(request.id.clone(), renewed.clone());
56                }
57                if let Some(record) = self.step_records.get_mut(&renewed.step_id) {
58                    record.renew_lease_until(renewed.expires_at);
59                }
60                self.updated_at = renewed.renewed_at;
61            }
62            CeremonyEvent::StepCompleted(completed) => self.apply_step_completed(completed),
63            CeremonyEvent::StepFailed(failed) => self.apply_step_failed(failed),
64            CeremonyEvent::ContextWritten(written) => self.apply_context_written(written),
65            CeremonyEvent::StateIterationStarted(started) => {
66                self.apply_state_iteration_started(started);
67            }
68            CeremonyEvent::TransitionApplied(applied) => self.apply_transition_applied(applied),
69            CeremonyEvent::InterventionRequested(requested) => {
70                self.apply_intervention_requested(requested);
71            }
72            CeremonyEvent::InterventionResponded(responded) => {
73                self.apply_intervention_responded(responded);
74            }
75            CeremonyEvent::InterventionClosed(closed) => self.apply_intervention_closed(closed),
76            CeremonyEvent::InterventionDeliveryAcknowledged(acknowledged) => {
77                self.apply_intervention_delivery_acknowledged(acknowledged);
78            }
79            CeremonyEvent::EvidenceCollected(collected) => self.apply_evidence_collected(collected),
80            CeremonyEvent::ReasonAsserted(asserted) => self.apply_reason_asserted(asserted),
81            CeremonyEvent::HumanApprovalRecorded(recorded) => {
82                self.apply_human_approval_recorded(recorded);
83            }
84            CeremonyEvent::HumanDeferralRecorded(recorded) => {
85                self.apply_human_deferral_recorded(recorded);
86            }
87            CeremonyEvent::CeremonyCompleted(completed) => self.apply_ceremony_completed(completed),
88            CeremonyEvent::InstanceImported(imported) => self.apply_instance_imported(imported),
89            CeremonyEvent::MemoryRecalled(recalled) => self.apply_memory_recalled(recalled),
90            CeremonyEvent::ChildSpawnPlanned(event) => self.apply_child_spawn_planned(event),
91            CeremonyEvent::ChildSpawnPlanAdopted(event) => {
92                self.apply_child_spawn_plan_adopted(event);
93            }
94            CeremonyEvent::ChildCompletionAccepted(event) => {
95                self.apply_child_completion_accepted(event);
96            }
97            CeremonyEvent::CeremonyPaused(event) => self.apply_ceremony_paused(event),
98            CeremonyEvent::CeremonyResumed(event) => self.apply_ceremony_resumed(event),
99            CeremonyEvent::CeremonyCancelled(event) => self.apply_ceremony_cancelled(event),
100            CeremonyEvent::CeremonyDeadlineExceeded(event) => {
101                self.apply_ceremony_deadline_exceeded(event);
102            }
103            CeremonyEvent::StateDeadlineExceeded(event) => {
104                self.apply_state_deadline_exceeded(event);
105            }
106            CeremonyEvent::StepDeadlineExceeded(event) => self.apply_step_deadline_exceeded(event),
107            CeremonyEvent::LateStepResultObserved(event) => {
108                self.apply_late_step_result_observed(event);
109            }
110            CeremonyEvent::ExecutionReceiptLinked(event) => {
111                self.apply_execution_receipt_linked(event);
112            }
113            CeremonyEvent::SuccessorPlanned(event) => self.apply_successor_planned(event),
114            CeremonyEvent::SuccessionCarried(event) => self.apply_succession_carried(event),
115        }
116    }
117
118    /// Fold a whole stream into the session it describes.
119    ///
120    /// The first event opens the session: either it was started here,
121    /// or it was imported from a store written before ceremonies were
122    /// streams (ADR-012). A stream that opens with anything else is not
123    /// a ceremony's stream and is refused, and so is one that carries
124    /// an import anywhere but at its first position — an import
125    /// replaces the whole session, so a second one would silently
126    /// discard everything between them.
127    pub fn rehydrate<'a>(
128        events: impl IntoIterator<Item = &'a CeremonyEvent>,
129    ) -> Result<Self, DomainError> {
130        let mut events = events.into_iter();
131        let mut instance = match events.next() {
132            Some(CeremonyEvent::CeremonyInstanceStarted(started)) => Self::from_started(started),
133            Some(CeremonyEvent::InstanceImported(imported)) => Self::from_imported(imported),
134            _ => {
135                return Err(DomainError::InvariantViolated {
136                    reason: "a ceremony stream opens with its start or with its import",
137                })
138            }
139        };
140        for event in events {
141            if matches!(event, CeremonyEvent::InstanceImported(_)) {
142                return Err(DomainError::InvariantViolated {
143                    reason: "a ceremony stream carries an import only as its first event",
144                });
145            }
146            instance.apply(event);
147        }
148        Ok(instance)
149    }
150
151    /// Fold the events one command decided, in order.
152    pub(in crate::entities::ceremony_instance) fn apply_all(&mut self, events: &[CeremonyEvent]) {
153        for event in events {
154            self.apply(event);
155        }
156    }
157}
made_core/entities/ceremony_instance/fold/mod.rs

made_core/entities/ceremony_instance/fold/
mod.rs
