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

ic_backup/workflow/fence_acquisition/
mod.rs
