ic_backup/workflow/ic_snapshot_transfer_read/
mod.rs

1//! One freshly reserved metadata/data update; no automatic receipt or retry.
2
3use crate::{
4    model::{
5        ic_snapshot_transfer_read::{
6            IcSnapshotTransferReadError, IcSnapshotTransferReadPayload,
7            IcSnapshotTransferReadRequest, IcSnapshotTransferReadResponse,
8        },
9        operation_plan::OperationPlanError,
10    },
11    ops::persistence::{
12        AttemptJournalError, AttemptJournalGuard, ExecutionProgressPersistenceError,
13        ExecutionStageGuard, ExecutionWorkflowPersistenceError,
14    },
15    policy::ic_snapshot_transfer_read::{
16        IcSnapshotTransferReadAssociationError, IcSnapshotTransferReadView, validate_response,
17    },
18    ports::{
19        ic_observation::IcObservationProviderError,
20        ic_snapshot_transfer_read::IcSnapshotTransferReadProvider,
21    },
22};
23use thiserror::Error;
24
25/// Durably reserve one original planned read, freshly admit it and invoke its provider once.
26///
27/// Hold no attempt guards on entry. The selected original journal remains locked
28/// through admission, dispatch and passive response association. Missing originals,
29/// pending attempts and unfulfilled prerequisites reject before provider invocation.
30/// This only accepts a newly reserved attempt; reconstructed pending requests are
31/// never dispatched. Every failure after reservation retains its consumed allowance.
32/// Admission and provider submission are awaited under that same selected guard.
33/// Dropping the future releases the guard without refunding or granting reentry.
34///
35/// `admit` must qualify actual fresh method-specific access, original metadata/raw-ID
36/// custody, current application requirements and exclusive never-dispatched command
37/// custody for this exact request. There is no default admission. Any remote preflight
38/// calls require their own prior accounting. The provider retains its existing
39/// authenticated single-update contract, without retries or hidden observations.
40///
41/// The returned bounded reply has only structural association. Integrations must
42/// independently authenticate it and explicitly use the existing journal receipt
43/// owner to record a qualified outcome. Even success leaves the attempt pending.
44/// No artifact, complete-transfer, fence or terminal/reference-release proof follows.
45/// # Errors
46/// Rejects changed originals, payloads, spending, fresh admission, provider failure
47/// or malformed/mismatched replies. Replies returned before a later rejection are
48/// retained in the error. No failure grants a repeat call, refund or cleanup.
49pub async fn read_snapshot<E: std::error::Error + 'static>(
50    stage: &ExecutionStageGuard<'_>,
51    operation_sequence: u64,
52    payload: IcSnapshotTransferReadPayload<'_, '_>,
53    provider: &mut impl IcSnapshotTransferReadProvider,
54    admit: impl AsyncFnOnce(&IcSnapshotTransferReadRequest<'_, '_>) -> Result<(), E>,
55) -> Result<IcSnapshotTransferReadResponse, IcSnapshotTransferReadExecutionError<E>> {
56    let (response, ()) = read_snapshot_with(
57        stage,
58        operation_sequence,
59        payload,
60        provider,
61        admit,
62        async |_, _, _| (),
63    )
64    .await?;
65    Ok(response)
66}
67
68// Complete a compound read while its original guard and decoded reply remain alive.
69// The continuation owns qualification and append/receipt ordering; checkpointing
70// happens only after this function has released the selected journal guard.
71pub(super) async fn read_snapshot_with<E: std::error::Error + 'static, T>(
72    stage: &ExecutionStageGuard<'_>,
73    operation_sequence: u64,
74    payload: IcSnapshotTransferReadPayload<'_, '_>,
75    provider: &mut impl IcSnapshotTransferReadProvider,
76    admit: impl AsyncFnOnce(&IcSnapshotTransferReadRequest<'_, '_>) -> Result<(), E>,
77    complete: impl AsyncFnOnce(
78        &mut AttemptJournalGuard<'_>,
79        &IcSnapshotTransferReadRequest<'_, '_>,
80        &IcSnapshotTransferReadView<'_, '_>,
81    ) -> T,
82) -> Result<(IcSnapshotTransferReadResponse, T), IcSnapshotTransferReadExecutionError<E>> {
83    let plan = stage.plan();
84    let authority = plan.attempt_authority(operation_sequence)?;
85    payload.validate_binding(&authority)?;
86    let mut journal = AttemptJournalGuard::open(stage.layout()?, &authority)?;
87    journal.reserve_planned_mutation(&plan.digest())?;
88    let request =
89        IcSnapshotTransferReadRequest::new(plan, operation_sequence, journal.record()?, payload)?;
90    let response = super::reserved_effect::submit_once(
91        &journal,
92        &request,
93        async |request| {
94            admit(request)
95                .await
96                .map_err(IcSnapshotTransferReadExecutionError::Admission)
97        },
98        || {
99            stage
100                .layout()
101                .map(|_| ())
102                .map_err(IcSnapshotTransferReadExecutionError::Stage)
103        },
104        async |request, record| {
105            provider
106                .read_snapshot(request, record)
107                .await
108                .map_err(IcSnapshotTransferReadExecutionError::Provider)
109        },
110        IcSnapshotTransferReadExecutionError::Journal,
111    )
112    .await?;
113    let admitted = match super::reserved_effect::check_returned(
114        stage,
115        &journal,
116        &request,
117        &response,
118        validate_response,
119    ) {
120        Ok((_, admitted)) => admitted,
121        Err(error) => {
122            return Err(match error {
123                super::reserved_effect::ReturnedError::Journal(source) => {
124                    IcSnapshotTransferReadExecutionError::AfterReplyJournal {
125                        source,
126                        response: Box::new(response),
127                    }
128                }
129                super::reserved_effect::ReturnedError::Association(source) => {
130                    IcSnapshotTransferReadExecutionError::Association {
131                        source,
132                        response: Box::new(response),
133                    }
134                }
135                super::reserved_effect::ReturnedError::Stage(source) => {
136                    IcSnapshotTransferReadExecutionError::AfterReplyStage {
137                        source,
138                        response: Box::new(response),
139                    }
140                }
141            });
142        }
143    };
144    let result = complete(&mut journal, &request, &admitted).await;
145    Ok((response, result))
146}
147
148/// Typed step failures retain original spending; returned raw replies remain available.
149#[derive(Debug, Error)]
150pub enum IcSnapshotTransferReadExecutionError<E: std::error::Error + 'static> {
151    /// The original operation cannot be derived.
152    #[error(transparent)]
153    Plan(#[from] OperationPlanError),
154    /// Retained workflow, stage or ancestors changed before dispatch.
155    #[error(transparent)]
156    Stage(#[from] ExecutionWorkflowPersistenceError),
157    /// Original journal admission or durable reservation failed.
158    #[error(transparent)]
159    Journal(#[from] AttemptJournalError),
160    /// Complete original progress, prerequisites or reservation failed.
161    #[error(transparent)]
162    Progress(#[from] ExecutionProgressPersistenceError),
163    /// Exact original payload/reservation admission failed.
164    #[error(transparent)]
165    Request(#[from] IcSnapshotTransferReadError),
166    /// Integration-owned fresh admission failed after durable reservation.
167    #[error("fresh snapshot read admission failed: {0}")]
168    Admission(#[source] E),
169    /// The single provider call failed; its reservation stays pending.
170    #[error(transparent)]
171    Provider(#[from] IcObservationProviderError),
172    /// Passive association rejected a bounded returned reply.
173    #[error("snapshot read response association failed: {source}")]
174    Association {
175        /// Canonical structural rejection.
176        source: IcSnapshotTransferReadAssociationError,
177        /// Exact bounded returned response, without authentication or outcome.
178        response: Box<IcSnapshotTransferReadResponse>,
179    },
180    /// Journal re-admission failed after a reply; retain the reply.
181    #[error("snapshot read journal changed after reply: {source}")]
182    AfterReplyJournal {
183        /// Original journal rejection.
184        source: AttemptJournalError,
185        /// Exact bounded returned response.
186        response: Box<IcSnapshotTransferReadResponse>,
187    },
188    /// Stage/ancestor re-admission failed after a reply; retain the reply.
189    #[error("snapshot read stage changed after reply: {source}")]
190    AfterReplyStage {
191        /// Original stage or ancestor rejection.
192        source: ExecutionWorkflowPersistenceError,
193        /// Exact bounded returned response.
194        response: Box<IcSnapshotTransferReadResponse>,
195    },
196}
197
198#[cfg(all(test, unix))]
199mod tests;
ic_backup/workflow/ic_snapshot_transfer_read/mod.rs

ic_backup/workflow/ic_snapshot_transfer_read/
mod.rs
