ic_backup/workflow/fence_acquisition/
mod.rs

1//! One freshly reserved application acquisition under exact original stage/obligation custody.
2
3use 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
22/// Reserve the original acquisition, freshly admit it and invoke one exact application update.
23///
24/// Hold no attempt guards on entry. Retain the full original stage, every journal,
25/// requirement and whole-unit obligation before calling. Restore requires its source
26/// layout/plan and outside-snapshot custody. Exact original payload admission precedes
27/// spending; complete original-plan admission owns the durable mutation reservation.
28/// The selected journal remains locked across admission and the provider await.
29/// Cancellation retains pending spending and denies redispatch through this step.
30///
31/// Mandatory fresh `admit` must qualify actual context, whole-unit membership,
32/// application method/argument purpose, original fence/revisions, permissions,
33/// dependencies, command custody and proof this reservation has never dispatched.
34/// Any remote preflight has its own prior accounting. The configured provider retains
35/// exact signed ingress before its one update and exact reply bytes before returning.
36/// There is no default application, permission or retention callback.
37///
38/// Recheck retained stage/ancestors and requirement/obligation around dispatch.
39/// Success returns passive association only and leaves the mutation pending.
40/// Applications independently qualify attribution and Active custody before explicit
41/// settlement; lost replies use original reconciliation accounting. No receipt, refund,
42/// automatic uncertainty, acquired fence or terminal/fence/reference release follows.
43/// # Errors
44/// Rejects changed/missing originals, binding/spending, fresh admission, provider or
45/// association failures. Every post-reservation error retains consumption; post-reply
46/// rejections retain the returned acknowledgement. No failure permits redispatch.
47pub 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    let acknowledgement = super::reserved_effect::submit_once(
64        &journal,
65        &request,
66        async |request| {
67            admit(request)
68                .await
69                .map_err(FenceAcquisitionExecutionError::Admission)
70        },
71        || {
72            read_fence_obligation(stage.layout()?, plan, requirement, &obligation.digest())?;
73            Ok(())
74        },
75        async |request, record| {
76            provider
77                .acquire_fence(request, record)
78                .await
79                .map_err(FenceAcquisitionExecutionError::Provider)
80        },
81        FenceAcquisitionExecutionError::Journal,
82    )
83    .await?;
84    let layout = match super::reserved_effect::check_returned(
85        stage,
86        &journal,
87        &request,
88        &acknowledgement,
89        |request, record, reply| validate_acknowledgement(request, record, reply).map(|_| ()),
90    ) {
91        Ok(layout) => layout,
92        Err(error) => {
93            return Err(match error {
94                super::reserved_effect::ReturnedError::Journal(source) => {
95                    FenceAcquisitionExecutionError::AfterReplyJournal {
96                        source,
97                        acknowledgement: Box::new(acknowledgement),
98                    }
99                }
100                super::reserved_effect::ReturnedError::Association(source) => {
101                    FenceAcquisitionExecutionError::Association {
102                        source,
103                        acknowledgement: Box::new(acknowledgement),
104                    }
105                }
106                super::reserved_effect::ReturnedError::Stage(source) => {
107                    FenceAcquisitionExecutionError::AfterReplyStage {
108                        source,
109                        acknowledgement: Box::new(acknowledgement),
110                    }
111                }
112            });
113        }
114    };
115    if let Err(source) = read_fence_obligation(layout, plan, requirement, &obligation.digest()) {
116        return Err(FenceAcquisitionExecutionError::AfterReplyObligation {
117            source,
118            acknowledgement: Box::new(acknowledgement),
119        });
120    }
121    Ok(acknowledgement)
122}
123
124/// Acquisition-step failure preserving original spending, obligations and passive replies.
125#[derive(Debug, Error)]
126pub enum FenceAcquisitionExecutionError<E: std::error::Error + 'static> {
127    /// Exact original acquisition authority cannot be derived.
128    #[error(transparent)]
129    Plan(#[from] OperationPlanError),
130    /// Original stage, workflow or ancestor evidence differs.
131    #[error(transparent)]
132    Stage(#[from] ExecutionWorkflowPersistenceError),
133    /// Original requirement or obligation cannot be admitted.
134    #[error(transparent)]
135    Obligation(#[from] FenceObligationPersistenceError),
136    /// Original selected journal admission failed.
137    #[error(transparent)]
138    Journal(#[from] AttemptJournalError),
139    /// Complete original journals, prerequisites or durable reservation failed.
140    #[error(transparent)]
141    Progress(#[from] ExecutionProgressPersistenceError),
142    /// Exact original payload or current reservation differs.
143    #[error(transparent)]
144    Request(#[from] FenceAcquisitionError),
145    /// Fresh application admission rejected after reservation.
146    #[error("fresh fence acquisition admission failed: {0}")]
147    Admission(#[source] E),
148    /// Single provider invocation failed; no nonapplication or refund is inferred.
149    #[error(transparent)]
150    Provider(#[from] FenceAcquisitionProviderError),
151    /// Passive original authority/attempt association rejected the reply.
152    #[error("fence acquisition acknowledgement association failed: {source}")]
153    Association {
154        /// Canonical passive association rejection.
155        source: FenceAcquisitionAcknowledgementError,
156        /// Exact returned passive acknowledgement, without acquisition proof.
157        acknowledgement: Box<FenceAcquisitionAcknowledgement>,
158    },
159    /// Selected journal re-admission failed after reply.
160    #[error("fence acquisition journal changed after reply: {source}")]
161    AfterReplyJournal {
162        /// Original journal rejection.
163        source: AttemptJournalError,
164        /// Exact returned passive acknowledgement.
165        acknowledgement: Box<FenceAcquisitionAcknowledgement>,
166    },
167    /// Retained stage/ancestor admission failed after reply.
168    #[error("fence acquisition stage changed after reply: {source}")]
169    AfterReplyStage {
170        /// Original stage/ancestor rejection.
171        source: ExecutionWorkflowPersistenceError,
172        /// Exact returned passive acknowledgement.
173        acknowledgement: Box<FenceAcquisitionAcknowledgement>,
174    },
175    /// Retained requirement/obligation admission failed after reply.
176    #[error("fence acquisition obligation changed after reply: {source}")]
177    AfterReplyObligation {
178        /// Original requirement/obligation rejection.
179        source: FenceObligationPersistenceError,
180        /// Exact returned passive acknowledgement.
181        acknowledgement: Box<FenceAcquisitionAcknowledgement>,
182    },
183}
184
185#[cfg(all(test, unix))]
186mod tests;
ic_backup/workflow/fence_acquisition/mod.rs

ic_backup/workflow/fence_acquisition/
mod.rs
