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 (response, result) = super::ic_mutation::execute_reserved(
44        stage,
45        &authority,
46        |_| Ok::<_, IcSnapshotTransferReadExecutionError<E>>(()),
47        |journal| {
48            Ok(IcSnapshotTransferReadRequest::new(
49                plan,
50                operation_sequence,
51                journal,
52                payload,
53            )?)
54        },
55        async |request| {
56            admit(request)
57                .await
58                .map_err(IcSnapshotTransferReadExecutionError::Admission)?;
59            stage.layout()?;
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    /// The original operation cannot be derived.
79    #[error(transparent)]
80    Plan(#[from] OperationPlanError),
81    /// Retained workflow, stage or ancestors changed before dispatch.
82    #[error(transparent)]
83    Stage(#[from] ExecutionWorkflowPersistenceError),
84    /// Original journal admission or durable reservation failed.
85    #[error(transparent)]
86    Journal(#[from] AttemptJournalError),
87    /// Complete original progress, prerequisites or reservation failed.
88    #[error(transparent)]
89    Progress(#[from] ExecutionProgressPersistenceError),
90    /// Exact original payload/reservation admission failed.
91    #[error(transparent)]
92    Request(#[from] IcSnapshotTransferReadError),
93    /// Integration-owned fresh admission failed after durable reservation.
94    #[error("fresh snapshot read admission failed: {0}")]
95    Admission(#[source] E),
96    /// The single provider call failed; its reservation stays pending.
97    #[error(transparent)]
98    Provider(#[from] IcObservationProviderError),
99    /// Admission after a reply failed, retaining its complete bounded evidence.
100    #[error("snapshot transfer read reply rejected: {source}")]
101    AfterReply {
102        /// Typed original admission failure.
103        source: IcSnapshotTransferReadReplyError,
104        /// Complete returned passive evidence, without settlement authority.
105        response: Box<IcSnapshotTransferReadResponse>,
106    },
107}
108
109/// Admission failure after a returned snapshot transfer read reply.
110#[derive(Debug, Error)]
111pub enum IcSnapshotTransferReadReplyError {
112    /// Original selected journal cannot be re-admitted.
113    #[error(transparent)]
114    Journal(#[from] AttemptJournalError),
115    /// Passive exact reply association failed.
116    #[error(transparent)]
117    Association(#[from] IcSnapshotTransferReadAssociationError),
118    /// Original stage or ancestor evidence cannot be re-admitted.
119    #[error(transparent)]
120    Stage(#[from] ExecutionWorkflowPersistenceError),
121}
122
123#[cfg(all(test, unix))]
124mod tests;
ic_backup/workflow/ic_snapshot_transfer_read/mod.rs

ic_backup/workflow/ic_snapshot_transfer_read/
mod.rs
