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