ic_backup/workflow/fence_reconciliation/
mod.rs1use crate::{
4 model::{
5 artifacts::ArtifactChecksumRecord,
6 attempt_journal::{ObservationOutcomeRecord, ObservationReceiptRequest},
7 fence_acquisition::{FenceAcquisitionError, FenceAcquisitionPayload},
8 fence_obligation::FenceObligationRecord,
9 fence_reconciliation::{
10 FenceReconciliationIntent, FenceReconciliationObservation, FenceReconciliationRequest,
11 FenceReconciliationRequestError,
12 },
13 operation_plan::OperationPlanError,
14 },
15 ops::persistence::{
16 AttemptJournalError, ExecutionProgressPersistenceError, ExecutionStageGuard,
17 ExecutionWorkflowPersistenceError, FenceObligationPersistenceError,
18 FenceObligationRequirement, read_fence_obligation,
19 },
20 policy::fence_reconciliation::{FenceReconciliationError, validate},
21 ports::fence_reconciliation::{FenceReconciliationProvider, FenceReconciliationProviderError},
22};
23use thiserror::Error;
24
25pub async fn observe_fence_acquisition<E: std::error::Error + 'static>(
51 stage: &ExecutionStageGuard<'_>,
52 requirement: FenceObligationRequirement<'_>,
53 obligation: &FenceObligationRecord,
54 payload: &FenceAcquisitionPayload,
55 challenge: ArtifactChecksumRecord,
56 provider: &mut impl FenceReconciliationProvider,
57 admit: impl AsyncFnOnce(&FenceReconciliationRequest<'_, '_>) -> Result<(), E>,
58) -> Result<FenceReconciliationObservation, FenceReconciliationExecutionError<E>> {
59 let plan = stage.plan();
60 let mut journal = super::fence_acquisition::open_original::<
61 FenceReconciliationExecutionError<E>,
62 >(stage, requirement, obligation, payload)?;
63 let intent = FenceReconciliationIntent::new(plan, obligation, journal.record()?, challenge)?;
64 journal.reserve_planned_observation(
65 &plan.digest(),
66 intent.mutation_attempt(),
67 intent.digest().hash(),
68 )?;
69 let request = FenceReconciliationRequest::new(&intent, journal.record()?)?;
70 admit(&request)
71 .await
72 .map_err(FenceReconciliationExecutionError::Admission)?;
73 read_fence_obligation(stage.layout()?, plan, requirement, &obligation.digest())?;
74 let observation = provider
75 .observe_fence_acquisition(&request, journal.record()?)
76 .await?;
77 let admit_reply = || -> Result<(), FenceReconciliationReplyError> {
78 validate(&request, journal.record()?, &observation)?;
79 read_fence_obligation(stage.layout()?, plan, requirement, &obligation.digest())?;
80 Ok(())
81 };
82 let admitted = admit_reply();
83 super::qualified_receipt::retain_reply(observation, admitted, |source, observation| {
84 FenceReconciliationExecutionError::AfterReply {
85 source,
86 observation,
87 }
88 })
89 .map(|(reply, ())| reply)
90}
91
92pub async fn settle_fence_acquisition_observation<E: std::error::Error + 'static>(
118 stage: &ExecutionStageGuard<'_>,
119 requirement: FenceObligationRequirement<'_>,
120 obligation: &FenceObligationRecord,
121 payload: &FenceAcquisitionPayload,
122 challenge: ArtifactChecksumRecord,
123 observation: &FenceReconciliationObservation,
124 qualify: impl AsyncFnOnce(
125 &FenceReconciliationRequest<'_, '_>,
126 &FenceReconciliationObservation,
127 ) -> Result<ObservationReceiptRequest, E>,
128) -> Result<ObservationOutcomeRecord, FenceReconciliationSettlementError<E>> {
129 let plan = stage.plan();
130 let mut journal = super::fence_acquisition::open_original::<
131 FenceReconciliationSettlementError<E>,
132 >(stage, requirement, obligation, payload)?;
133 journal.admit_planned_progress(&plan.digest())?;
134 let intent = FenceReconciliationIntent::new(plan, obligation, journal.record()?, challenge)?;
135 let request = FenceReconciliationRequest::new(&intent, journal.record()?)?;
136 let admitted = validate(&request, journal.record()?, observation)?;
137 let receipt = qualify(&request, observation)
138 .await
139 .map_err(FenceReconciliationSettlementError::Qualification)?;
140 if receipt.attempt != admitted.observation_attempt()
141 || receipt.request != admitted.request().hash()
142 || receipt.outcome != admitted.outcome()
143 || receipt.evidence != admitted.evidence().hash()
144 {
145 return Err(FenceReconciliationSettlementError::ReceiptMismatch);
146 }
147 validate(&request, journal.record()?, observation)?;
148 read_fence_obligation(stage.layout()?, plan, requirement, &obligation.digest())?;
149 let outcome = receipt.outcome;
150 journal.record_planned_observation(&plan.digest(), receipt)?;
151 read_fence_obligation(stage.layout()?, plan, requirement, &obligation.digest())?;
152 Ok(outcome)
153}
154
155#[derive(Debug, Error)]
157pub enum FenceReconciliationSettlementError<E: std::error::Error + 'static> {
158 #[error(transparent)]
160 Plan(#[from] OperationPlanError),
161 #[error(transparent)]
163 Stage(#[from] ExecutionWorkflowPersistenceError),
164 #[error(transparent)]
166 Obligation(#[from] FenceObligationPersistenceError),
167 #[error(transparent)]
169 Journal(#[from] AttemptJournalError),
170 #[error(transparent)]
172 Payload(#[from] FenceAcquisitionError),
173 #[error(transparent)]
175 Progress(#[from] ExecutionProgressPersistenceError),
176 #[error(transparent)]
178 Request(#[from] FenceReconciliationRequestError),
179 #[error(transparent)]
181 Association(#[from] FenceReconciliationError),
182 #[error("fence observation settlement qualification failed: {0}")]
184 Qualification(#[source] E),
185 #[error("fence observation settlement receipt mismatch")]
187 ReceiptMismatch,
188}
189
190#[derive(Debug, Error)]
192pub enum FenceReconciliationExecutionError<E: std::error::Error + 'static> {
193 #[error(transparent)]
195 Plan(#[from] OperationPlanError),
196 #[error(transparent)]
198 Stage(#[from] ExecutionWorkflowPersistenceError),
199 #[error(transparent)]
201 Obligation(#[from] FenceObligationPersistenceError),
202 #[error(transparent)]
204 Journal(#[from] AttemptJournalError),
205 #[error(transparent)]
207 Payload(#[from] FenceAcquisitionError),
208 #[error(transparent)]
210 Progress(#[from] ExecutionProgressPersistenceError),
211 #[error(transparent)]
213 Request(#[from] FenceReconciliationRequestError),
214 #[error("fresh fence reconciliation admission failed: {0}")]
216 Admission(#[source] E),
217 #[error(transparent)]
219 Provider(#[from] FenceReconciliationProviderError),
220 #[error("fence reconciliation reply rejected: {source}")]
222 AfterReply {
223 source: FenceReconciliationReplyError,
225 observation: Box<FenceReconciliationObservation>,
227 },
228}
229
230#[derive(Debug, Error)]
232pub enum FenceReconciliationReplyError {
233 #[error(transparent)]
235 Journal(#[from] AttemptJournalError),
236 #[error(transparent)]
238 Association(#[from] FenceReconciliationError),
239 #[error(transparent)]
241 Stage(#[from] ExecutionWorkflowPersistenceError),
242 #[error(transparent)]
244 Obligation(#[from] FenceObligationPersistenceError),
245}