made_core/entities/ceremony_instance/
step_execution.rs

1use super::{
2    CeremonyDefinition, CeremonyInstance, DomainError, OffsetDateTime, RoleAction, RoleId,
3    StepAttempt, StepExecutionRecord, StepId, StepLease, StepResult, StepStatus,
4};
5
6impl CeremonyInstance {
7    pub fn start_step_as(
8        &mut self,
9        definition: &CeremonyDefinition,
10        role_id: &RoleId,
11        step_id: &StepId,
12        lease: StepLease,
13        now: OffsetDateTime,
14    ) -> Result<StepAttempt, DomainError> {
15        self.require_role(definition, role_id, &RoleAction::step(step_id.clone()))?;
16        self.start_step(definition, step_id, lease, now)
17    }
18
19    pub fn start_step(
20        &mut self,
21        definition: &CeremonyDefinition,
22        step_id: &StepId,
23        lease: StepLease,
24        now: OffsetDateTime,
25    ) -> Result<StepAttempt, DomainError> {
26        self.require_definition(definition)?;
27        if self.is_terminal(definition) {
28            return Err(DomainError::InvariantViolated {
29                reason: "terminal ceremony instances cannot start steps",
30            });
31        }
32
33        let step = definition.step(step_id).ok_or(DomainError::NotFound {
34            what: "ceremony_instance.step",
35        })?;
36        if step.state_id() != &self.current_state {
37            return Err(DomainError::InvalidTransition {
38                from: "ceremony_instance.current_state",
39                to: "ceremony_step.state",
40            });
41        }
42
43        let record = self
44            .step_records
45            .get(step_id)
46            .cloned()
47            .ok_or(DomainError::NotFound {
48                what: "ceremony_instance.step_record",
49            })?;
50        if !record.can_be_started_at(now) {
51            return Err(DomainError::InvariantViolated {
52                reason: "step lease is still active",
53            });
54        }
55
56        let next_attempt = next_attempt_for_start(&record)?;
57        if !step.retry_policy().allows_attempt(next_attempt) {
58            return Err(DomainError::InvariantViolated {
59                reason: "step retry policy exhausted",
60            });
61        }
62        if !self
63            .idempotency_keys
64            .insert(lease.idempotency_key().clone())
65        {
66            return Err(DomainError::AlreadyExists {
67                what: "ceremony_instance.idempotency_key",
68            });
69        }
70
71        self.step_records
72            .insert(step_id.clone(), record.with_started(lease, next_attempt));
73        self.updated_at = now;
74        Ok(next_attempt)
75    }
76
77    pub fn apply_step_result(
78        &mut self,
79        definition: &CeremonyDefinition,
80        step_id: &StepId,
81        result: StepResult,
82        now: OffsetDateTime,
83    ) -> Result<(), DomainError> {
84        self.require_definition(definition)?;
85        let step = definition.step(step_id).ok_or(DomainError::NotFound {
86            what: "ceremony_instance.step",
87        })?;
88        if step.state_id() != &self.current_state {
89            return Err(DomainError::InvalidTransition {
90                from: "ceremony_instance.current_state",
91                to: "ceremony_step.state",
92            });
93        }
94
95        let record = self
96            .step_records
97            .get(step_id)
98            .cloned()
99            .ok_or(DomainError::NotFound {
100                what: "ceremony_instance.step_record",
101            })?;
102        if record.status() != StepStatus::InProgress {
103            return Err(DomainError::InvariantViolated {
104                reason: "step result requires an in-progress step",
105            });
106        }
107
108        let finished = record.with_result(result);
109        let repeat = step.repeat_policy().filter(|policy| {
110            finished.status().is_success() && !policy.is_satisfied(finished.output())
111        });
112        if repeat.is_some_and(|policy| policy.permits_another_iteration(finished.iteration())) {
113            let next_iteration = finished.iteration().next()?;
114            self.step_record_history
115                .entry(step_id.clone())
116                .or_default()
117                .push(finished);
118            self.step_records.insert(
119                step_id.clone(),
120                StepExecutionRecord::pending_iteration(next_iteration),
121            );
122        } else {
123            self.step_records.insert(step_id.clone(), finished);
124        }
125        self.updated_at = now;
126        Ok(())
127    }
128}
129
130fn next_attempt_for_start(record: &StepExecutionRecord) -> Result<StepAttempt, DomainError> {
131    if matches!(record.status(), StepStatus::Failed | StepStatus::InProgress) {
132        record.attempt().next()
133    } else {
134        Ok(record.attempt())
135    }
136}
made_core/entities/ceremony_instance/step_execution.rs

made_core/entities/ceremony_instance/
step_execution.rs
