ic_backup/workflow/fence_acquisition/
mod.rs1use crate::{
4 model::{
5 fence_acquisition::{
6 FenceAcquisitionAcknowledgement, FenceAcquisitionError, FenceAcquisitionPayload,
7 FenceAcquisitionRequest, original_authority,
8 },
9 fence_obligation::FenceObligationRecord,
10 operation_plan::OperationPlanError,
11 },
12 ops::persistence::{
13 AttemptJournalError, AttemptJournalGuard, ExecutionProgressPersistenceError,
14 ExecutionStageGuard, ExecutionWorkflowPersistenceError, FenceObligationPersistenceError,
15 FenceObligationRequirement, read_fence_obligation,
16 },
17 policy::fence_acquisition::{FenceAcquisitionAcknowledgementError, validate_acknowledgement},
18 ports::fence_acquisition::{FenceAcquisitionProvider, FenceAcquisitionProviderError},
19};
20use thiserror::Error;
21
22pub async fn acquire_fence<E: std::error::Error + 'static>(
48 stage: &ExecutionStageGuard<'_>,
49 requirement: FenceObligationRequirement<'_>,
50 obligation: &FenceObligationRecord,
51 payload: &FenceAcquisitionPayload,
52 provider: &mut impl FenceAcquisitionProvider,
53 admit: impl AsyncFnOnce(&FenceAcquisitionRequest<'_>) -> Result<(), E>,
54) -> Result<FenceAcquisitionAcknowledgement, FenceAcquisitionExecutionError<E>> {
55 let plan = stage.plan();
56 let layout = stage.layout()?;
57 read_fence_obligation(layout, plan, requirement, &obligation.digest())?;
58 let authority = plan.attempt_authority(obligation.acquisition_operation())?;
59 let mut journal = AttemptJournalGuard::open(layout, &authority)?;
60 original_authority(plan, obligation, journal.record()?, payload)?;
61 journal.reserve_planned_mutation(&plan.digest())?;
62 let request = FenceAcquisitionRequest::new(plan, obligation, journal.record()?, payload)?;
63 admit(&request)
64 .await
65 .map_err(FenceAcquisitionExecutionError::Admission)?;
66 read_fence_obligation(stage.layout()?, plan, requirement, &obligation.digest())?;
67 let acknowledgement = provider.acquire_fence(&request, journal.record()?).await?;
68 let admit_reply = || -> Result<(), FenceAcquisitionReplyError> {
69 validate_acknowledgement(&request, journal.record()?, &acknowledgement)?;
70 read_fence_obligation(stage.layout()?, plan, requirement, &obligation.digest())?;
71 Ok(())
72 };
73 match admit_reply() {
74 Ok(()) => Ok(acknowledgement),
75 Err(source) => Err(FenceAcquisitionExecutionError::AfterReply {
76 source,
77 acknowledgement: Box::new(acknowledgement),
78 }),
79 }
80}
81
82#[derive(Debug, Error)]
84pub enum FenceAcquisitionExecutionError<E: std::error::Error + 'static> {
85 #[error(transparent)]
87 Plan(#[from] OperationPlanError),
88 #[error(transparent)]
90 Stage(#[from] ExecutionWorkflowPersistenceError),
91 #[error(transparent)]
93 Obligation(#[from] FenceObligationPersistenceError),
94 #[error(transparent)]
96 Journal(#[from] AttemptJournalError),
97 #[error(transparent)]
99 Progress(#[from] ExecutionProgressPersistenceError),
100 #[error(transparent)]
102 Request(#[from] FenceAcquisitionError),
103 #[error("fresh fence acquisition admission failed: {0}")]
105 Admission(#[source] E),
106 #[error(transparent)]
108 Provider(#[from] FenceAcquisitionProviderError),
109 #[error("fence acquisition reply rejected: {source}")]
111 AfterReply {
112 source: FenceAcquisitionReplyError,
114 acknowledgement: Box<FenceAcquisitionAcknowledgement>,
116 },
117}
118
119#[derive(Debug, Error)]
121pub enum FenceAcquisitionReplyError {
122 #[error(transparent)]
124 Journal(#[from] AttemptJournalError),
125 #[error(transparent)]
127 Association(#[from] FenceAcquisitionAcknowledgementError),
128 #[error(transparent)]
130 Stage(#[from] ExecutionWorkflowPersistenceError),
131 #[error(transparent)]
133 Obligation(#[from] FenceObligationPersistenceError),
134}
135
136#[cfg(all(test, unix))]
137mod tests;