ic_backup/workflow/ic_snapshot_upload/
mod.rs

1//! One fresh original source-bound metadata/data upload; no automatic receipt or retry.
2
3mod allocation;
4#[cfg(unix)]
5mod data;
6#[cfg(unix)]
7pub use data::{
8    IcSnapshotDataUploadExecutionError, IcSnapshotDataUploadReplyError, upload_snapshot_data,
9};
10
11pub use allocation::{
12    IcSnapshotAllocationExecutionError, IcSnapshotAllocationSettlementError, allocate_snapshot,
13};
14
15use crate::ports::IcProviderError;
16use crate::{
17    model::{
18        ic_mutation::IcMutationAcknowledgement,
19        ic_snapshot_upload::{
20            IcSnapshotUploadAttempt, IcSnapshotUploadAttemptError, IcSnapshotUploadRequest,
21            original_authority,
22        },
23        operation_plan::OperationPlanError,
24    },
25    ops::persistence::{
26        AttemptJournalError, AttemptJournalGuard, ExecutionStageGuard,
27        ExecutionWorkflowPersistenceError,
28    },
29    policy::ic_snapshot_upload::{
30        IcSnapshotUploadAcknowledgementView, IcSnapshotUploadAssociationError,
31        validate_acknowledgement,
32    },
33    ports::ic_snapshot_upload::IcSnapshotUploadProvider,
34};
35use thiserror::Error;
36
37// Complete compound uploads under their original guard using the admitted reply.
38// Return before checkpointing, which must run after journal exclusion is released.
39async fn upload_snapshot_with<E: std::error::Error + 'static, T>(
40    stage: &ExecutionStageGuard<'_>,
41    operation_sequence: u64,
42    payload: &IcSnapshotUploadRequest<'_>,
43    provider: &mut impl IcSnapshotUploadProvider,
44    admit: impl AsyncFnOnce(&IcSnapshotUploadAttempt<'_, '_>) -> Result<(), E>,
45    complete: impl AsyncFnOnce(
46        &mut AttemptJournalGuard<'_>,
47        &IcSnapshotUploadAttempt<'_, '_>,
48        &IcSnapshotUploadAcknowledgementView<'_, '_>,
49    ) -> T,
50) -> Result<(IcMutationAcknowledgement, T), IcSnapshotUploadExecutionError<E>> {
51    let plan = stage.plan();
52    let authority = plan.attempt_authority(operation_sequence)?;
53    let (acknowledgement, result) = super::ic_mutation::execute_reserved(
54        stage,
55        &authority,
56        |journal| {
57            original_authority(plan, operation_sequence, journal, payload)?;
58            Ok::<_, IcSnapshotUploadExecutionError<E>>(())
59        },
60        |journal| {
61            Ok(IcSnapshotUploadAttempt::new(
62                plan,
63                operation_sequence,
64                journal,
65                payload,
66            )?)
67        },
68        async |request| {
69            admit(request)
70                .await
71                .map_err(IcSnapshotUploadExecutionError::Admission)?;
72            stage
73                .layout()
74                .map_err(super::ExecutionReservationError::from)?;
75            Ok(())
76        },
77        async |request, journal| Ok(provider.submit_upload(request, journal).await?),
78        async |journal, request, acknowledgement| {
79            let admitted = validate_acknowledgement(request, journal.record()?, acknowledgement)?;
80            stage.layout()?;
81            Ok::<_, IcSnapshotUploadReplyError>(complete(journal, request, &admitted).await)
82        },
83    )
84    .await?;
85    super::qualified_receipt::retain_reply(acknowledgement, result, |source, acknowledgement| {
86        IcSnapshotUploadExecutionError::AfterReply {
87            source,
88            acknowledgement,
89        }
90    })
91}
92
93/// Upload-step failures retain original spending and any bounded returned acknowledgement.
94#[derive(Debug, Error)]
95pub enum IcSnapshotUploadExecutionError<E: std::error::Error + 'static> {
96    /// Shared original-stage, journal or durable reservation admission failed.
97    #[error(transparent)]
98    Reservation(#[from] super::ExecutionReservationError),
99    /// Original operation authority cannot be derived.
100    #[error(transparent)]
101    Plan(#[from] OperationPlanError),
102    /// Canonical original mutation binding or current reservation differs.
103    #[error(transparent)]
104    Request(#[from] IcSnapshotUploadAttemptError),
105    /// Fresh integration-owned admission rejected after reservation.
106    #[error("fresh snapshot upload admission failed: {0}")]
107    Admission(#[source] E),
108    /// The single provider call failed; spending remains pending.
109    #[error(transparent)]
110    Provider(#[from] IcProviderError),
111    /// Admission after a reply failed, retaining its complete bounded evidence.
112    #[error("snapshot upload reply rejected: {source}")]
113    AfterReply {
114        /// Typed original admission failure.
115        source: IcSnapshotUploadReplyError,
116        /// Complete returned passive evidence, without settlement authority.
117        acknowledgement: Box<IcMutationAcknowledgement>,
118    },
119}
120
121/// Admission failure after a returned snapshot upload reply.
122#[derive(Debug, Error)]
123pub enum IcSnapshotUploadReplyError {
124    /// Original selected journal cannot be re-admitted.
125    #[error(transparent)]
126    Journal(#[from] AttemptJournalError),
127    /// Passive exact reply association failed.
128    #[error(transparent)]
129    Association(#[from] IcSnapshotUploadAssociationError),
130    /// Original stage or ancestor evidence cannot be re-admitted.
131    #[error(transparent)]
132    Stage(#[from] ExecutionWorkflowPersistenceError),
133}
134
135impl<E: std::error::Error + 'static> IcSnapshotUploadExecutionError<E> {
136    /// Borrow the bounded returned evidence carried by this error, including nested failures.
137    /// Absence describes retained evidence only and grants no dispatch or retry authority.
138    #[must_use]
139    pub fn acknowledgement(&self) -> Option<&IcMutationAcknowledgement> {
140        match self {
141            Self::AfterReply {
142                acknowledgement, ..
143            } => Some(acknowledgement),
144            _ => None,
145        }
146    }
147}
ic_backup/workflow/ic_snapshot_upload/mod.rs

ic_backup/workflow/ic_snapshot_upload/
mod.rs
