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// Complete a compound read while its original guard and decoded reply remain alive.
26// The continuation owns qualification and append/receipt ordering; checkpointing
27// happens only after this function has released the selected journal guard.
28pub(super) async fn read_snapshot_with<E: std::error::Error + 'static, T>(
29    stage: &ExecutionStageGuard<'_>,
30    operation_sequence: u64,
31    payload: IcSnapshotTransferReadPayload<'_, '_>,
32    provider: &mut impl IcSnapshotTransferReadProvider,
33    admit: impl AsyncFnOnce(&IcSnapshotTransferReadRequest<'_, '_>) -> Result<(), E>,
34    complete: impl AsyncFnOnce(
35        &mut AttemptJournalGuard<'_>,
36        &IcSnapshotTransferReadRequest<'_, '_>,
37        &IcSnapshotTransferReadView<'_, '_>,
38    ) -> T,
39) -> Result<(IcSnapshotTransferReadResponse, T), IcSnapshotTransferReadExecutionError<E>> {
40    let plan = stage.plan();
41    let authority = plan.attempt_authority(operation_sequence)?;
42    payload.validate_binding(&authority)?;
43    let mut journal = AttemptJournalGuard::open(stage.layout()?, &authority)?;
44    journal.reserve_planned_mutation(&plan.digest())?;
45    let request =
46        IcSnapshotTransferReadRequest::new(plan, operation_sequence, journal.record()?, payload)?;
47    admit(&request)
48        .await
49        .map_err(IcSnapshotTransferReadExecutionError::Admission)?;
50    stage.layout()?;
51    let response = provider.read_snapshot(&request, journal.record()?).await?;
52    let admit_reply = || {
53        let admitted = validate_response(&request, journal.record()?, &response)?;
54        stage.layout()?;
55        Ok::<_, IcSnapshotTransferReadReplyError>(admitted)
56    };
57    let admitted = match admit_reply() {
58        Ok(admitted) => admitted,
59        Err(source) => {
60            return Err(IcSnapshotTransferReadExecutionError::AfterReply {
61                source,
62                response: Box::new(response),
63            });
64        }
65    };
66    let result = complete(&mut journal, &request, &admitted).await;
67    Ok((response, result))
68}
69
70/// Typed step failures retain original spending; returned raw replies remain available.
71#[derive(Debug, Error)]
72pub enum IcSnapshotTransferReadExecutionError<E: std::error::Error + 'static> {
73    /// The original operation cannot be derived.
74    #[error(transparent)]
75    Plan(#[from] OperationPlanError),
76    /// Retained workflow, stage or ancestors changed before dispatch.
77    #[error(transparent)]
78    Stage(#[from] ExecutionWorkflowPersistenceError),
79    /// Original journal admission or durable reservation failed.
80    #[error(transparent)]
81    Journal(#[from] AttemptJournalError),
82    /// Complete original progress, prerequisites or reservation failed.
83    #[error(transparent)]
84    Progress(#[from] ExecutionProgressPersistenceError),
85    /// Exact original payload/reservation admission failed.
86    #[error(transparent)]
87    Request(#[from] IcSnapshotTransferReadError),
88    /// Integration-owned fresh admission failed after durable reservation.
89    #[error("fresh snapshot read admission failed: {0}")]
90    Admission(#[source] E),
91    /// The single provider call failed; its reservation stays pending.
92    #[error(transparent)]
93    Provider(#[from] IcObservationProviderError),
94    /// Admission after a reply failed, retaining its complete bounded evidence.
95    #[error("snapshot transfer read reply rejected: {source}")]
96    AfterReply {
97        /// Typed original admission failure.
98        source: IcSnapshotTransferReadReplyError,
99        /// Complete returned passive evidence, without settlement authority.
100        response: Box<IcSnapshotTransferReadResponse>,
101    },
102}
103
104/// Admission failure after a returned snapshot transfer read reply.
105#[derive(Debug, Error)]
106pub enum IcSnapshotTransferReadReplyError {
107    /// Original selected journal cannot be re-admitted.
108    #[error(transparent)]
109    Journal(#[from] AttemptJournalError),
110    /// Passive exact reply association failed.
111    #[error(transparent)]
112    Association(#[from] IcSnapshotTransferReadAssociationError),
113    /// Original stage or ancestor evidence cannot be re-admitted.
114    #[error(transparent)]
115    Stage(#[from] ExecutionWorkflowPersistenceError),
116}
117
118#[cfg(all(test, unix))]
119mod tests;
ic_backup/workflow/ic_snapshot_transfer_read/mod.rs

ic_backup/workflow/ic_snapshot_transfer_read/
mod.rs
