ic_backup/workflow/ic_snapshot_transfer_read/
mod.rs

1//! One freshly reserved metadata/data update; no automatic receipt or retry.
2
3use crate::ports::IcProviderError;
4use crate::{
5    model::{
6        ic_snapshot_transfer_read::{
7            IcSnapshotTransferReadError, IcSnapshotTransferReadPayload,
8            IcSnapshotTransferReadRequest, IcSnapshotTransferReadResponse,
9        },
10        operation_plan::OperationPlanError,
11    },
12    ops::persistence::{
13        AttemptJournalError, AttemptJournalGuard, ExecutionStageGuard,
14        ExecutionWorkflowPersistenceError,
15    },
16    policy::ic_snapshot_transfer_read::{
17        IcSnapshotTransferReadAssociationError, IcSnapshotTransferReadView, validate_response,
18    },
19    ports::ic_snapshot_transfer_read::IcSnapshotTransferReadProvider,
20};
21use thiserror::Error;
22
23// Complete a compound read while its original guard and decoded reply remain alive.
24// The continuation owns qualification and append/receipt ordering; checkpointing
25// happens only after this function has released the selected journal guard.
26pub(super) async fn read_snapshot_with<E: std::error::Error + 'static, T>(
27    stage: &ExecutionStageGuard<'_>,
28    operation_sequence: u64,
29    payload: IcSnapshotTransferReadPayload<'_, '_>,
30    provider: &mut impl IcSnapshotTransferReadProvider,
31    admit: impl AsyncFnOnce(&IcSnapshotTransferReadRequest<'_, '_>) -> Result<(), E>,
32    complete: impl AsyncFnOnce(
33        &mut AttemptJournalGuard<'_>,
34        &IcSnapshotTransferReadRequest<'_, '_>,
35        &IcSnapshotTransferReadView<'_, '_>,
36    ) -> T,
37) -> Result<(IcSnapshotTransferReadResponse, T), IcSnapshotTransferReadExecutionError<E>> {
38    let plan = stage.plan();
39    let authority = plan.attempt_authority(operation_sequence)?;
40    payload.validate_binding(&authority)?;
41    let (response, result) = super::ic_mutation::execute_reserved(
42        stage,
43        &authority,
44        |_| Ok::<_, IcSnapshotTransferReadExecutionError<E>>(()),
45        |journal| {
46            Ok(IcSnapshotTransferReadRequest::new(
47                plan,
48                operation_sequence,
49                journal,
50                payload,
51            )?)
52        },
53        async |request| {
54            admit(request)
55                .await
56                .map_err(IcSnapshotTransferReadExecutionError::Admission)?;
57            stage
58                .layout()
59                .map_err(super::ExecutionReservationError::from)?;
60            Ok(())
61        },
62        async |request, journal| Ok(provider.read_snapshot(request, journal).await?),
63        async |journal, request, response| {
64            let admitted = validate_response(request, journal.record()?, response)?;
65            stage.layout()?;
66            Ok::<_, IcSnapshotTransferReadReplyError>(complete(journal, request, &admitted).await)
67        },
68    )
69    .await?;
70    super::qualified_receipt::retain_reply(response, result, |source, response| {
71        IcSnapshotTransferReadExecutionError::AfterReply { source, response }
72    })
73}
74
75/// Typed step failures retain original spending; returned raw replies remain available.
76#[derive(Debug, Error)]
77pub enum IcSnapshotTransferReadExecutionError<E: std::error::Error + 'static> {
78    /// Shared original-stage, journal or durable reservation admission failed.
79    #[error(transparent)]
80    Reservation(#[from] super::ExecutionReservationError),
81    /// The original operation cannot be derived.
82    #[error(transparent)]
83    Plan(#[from] OperationPlanError),
84    /// Exact original payload/reservation admission failed.
85    #[error(transparent)]
86    Request(#[from] IcSnapshotTransferReadError),
87    /// Integration-owned fresh admission failed after durable reservation.
88    #[error("fresh snapshot read admission failed: {0}")]
89    Admission(#[source] E),
90    /// The single provider call failed; its reservation stays pending.
91    #[error(transparent)]
92    Provider(#[from] IcProviderError),
93    /// Admission after a reply failed, retaining its complete bounded evidence.
94    #[error("snapshot transfer read reply rejected: {source}")]
95    AfterReply {
96        /// Typed original admission failure.
97        source: IcSnapshotTransferReadReplyError,
98        /// Complete returned passive evidence, without settlement authority.
99        response: Box<IcSnapshotTransferReadResponse>,
100    },
101}
102
103/// Admission failure after a returned snapshot transfer read reply.
104#[derive(Debug, Error)]
105pub enum IcSnapshotTransferReadReplyError {
106    /// Original selected journal cannot be re-admitted.
107    #[error(transparent)]
108    Journal(#[from] AttemptJournalError),
109    /// Passive exact reply association failed.
110    #[error(transparent)]
111    Association(#[from] IcSnapshotTransferReadAssociationError),
112    /// Original stage or ancestor evidence cannot be re-admitted.
113    #[error(transparent)]
114    Stage(#[from] ExecutionWorkflowPersistenceError),
115}
116
117impl<E: std::error::Error + 'static> IcSnapshotTransferReadExecutionError<E> {
118    /// Borrow the bounded returned evidence carried by this error, including nested failures.
119    /// Absence describes retained evidence only and grants no dispatch or retry authority.
120    #[must_use]
121    pub fn response(&self) -> Option<&IcSnapshotTransferReadResponse> {
122        match self {
123            Self::AfterReply { response, .. } => Some(response),
124            _ => None,
125        }
126    }
127}
ic_backup/workflow/ic_snapshot_transfer_read/mod.rs

ic_backup/workflow/ic_snapshot_transfer_read/
mod.rs
