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::{
16    model::{
17        ic_mutation::IcMutationAcknowledgement,
18        ic_snapshot_upload::{
19            IcSnapshotUploadAttempt, IcSnapshotUploadAttemptError, IcSnapshotUploadRequest,
20            original_authority,
21        },
22        operation_plan::OperationPlanError,
23    },
24    ops::persistence::{
25        AttemptJournalError, AttemptJournalGuard, ExecutionProgressPersistenceError,
26        ExecutionStageGuard, ExecutionWorkflowPersistenceError,
27    },
28    policy::ic_snapshot_upload::{
29        IcSnapshotUploadAcknowledgementView, IcSnapshotUploadAssociationError,
30        validate_acknowledgement,
31    },
32    ports::{ic_mutation::IcMutationProviderError, ic_snapshot_upload::IcSnapshotUploadProvider},
33};
34use thiserror::Error;
35
36// Complete compound uploads under their original guard using the admitted reply.
37// Return before checkpointing, which must run after journal exclusion is released.
38async fn upload_snapshot_with<E: std::error::Error + 'static, T>(
39    stage: &ExecutionStageGuard<'_>,
40    operation_sequence: u64,
41    payload: &IcSnapshotUploadRequest<'_>,
42    provider: &mut impl IcSnapshotUploadProvider,
43    admit: impl AsyncFnOnce(&IcSnapshotUploadAttempt<'_, '_>) -> Result<(), E>,
44    complete: impl AsyncFnOnce(
45        &mut AttemptJournalGuard<'_>,
46        &IcSnapshotUploadAttempt<'_, '_>,
47        &IcSnapshotUploadAcknowledgementView<'_, '_>,
48    ) -> T,
49) -> Result<(IcMutationAcknowledgement, T), IcSnapshotUploadExecutionError<E>> {
50    let plan = stage.plan();
51    let authority = plan.attempt_authority(operation_sequence)?;
52    let mut journal = AttemptJournalGuard::open(stage.layout()?, &authority)?;
53    original_authority(plan, operation_sequence, journal.record()?, payload)?;
54    journal.reserve_planned_mutation(&plan.digest())?;
55    let request =
56        IcSnapshotUploadAttempt::new(plan, operation_sequence, journal.record()?, payload)?;
57    admit(&request)
58        .await
59        .map_err(IcSnapshotUploadExecutionError::Admission)?;
60    stage.layout()?;
61    let acknowledgement = provider.submit_upload(&request, journal.record()?).await?;
62    let admit_reply = || {
63        let admitted = validate_acknowledgement(&request, journal.record()?, &acknowledgement)?;
64        stage.layout()?;
65        Ok::<_, IcSnapshotUploadReplyError>(admitted)
66    };
67    let admitted = match admit_reply() {
68        Ok(admitted) => admitted,
69        Err(source) => {
70            return Err(IcSnapshotUploadExecutionError::AfterReply {
71                source,
72                acknowledgement: Box::new(acknowledgement),
73            });
74        }
75    };
76    let result = complete(&mut journal, &request, &admitted).await;
77    Ok((acknowledgement, result))
78}
79
80/// Upload-step failures retain original spending and any bounded returned acknowledgement.
81#[derive(Debug, Error)]
82pub enum IcSnapshotUploadExecutionError<E: std::error::Error + 'static> {
83    /// Original operation authority cannot be derived.
84    #[error(transparent)]
85    Plan(#[from] OperationPlanError),
86    /// Workflow, stage or original ancestor evidence failed admission.
87    #[error(transparent)]
88    Stage(#[from] ExecutionWorkflowPersistenceError),
89    /// Original selected journal admission failed.
90    #[error(transparent)]
91    Journal(#[from] AttemptJournalError),
92    /// Complete original progress, prerequisites or durable reservation failed.
93    #[error(transparent)]
94    Progress(#[from] ExecutionProgressPersistenceError),
95    /// Canonical original mutation binding or current reservation differs.
96    #[error(transparent)]
97    Request(#[from] IcSnapshotUploadAttemptError),
98    /// Fresh integration-owned admission rejected after reservation.
99    #[error("fresh snapshot upload admission failed: {0}")]
100    Admission(#[source] E),
101    /// The single provider call failed; spending remains pending.
102    #[error(transparent)]
103    Provider(#[from] IcMutationProviderError),
104    /// Admission after a reply failed, retaining its complete bounded evidence.
105    #[error("snapshot upload reply rejected: {source}")]
106    AfterReply {
107        /// Typed original admission failure.
108        source: IcSnapshotUploadReplyError,
109        /// Complete returned passive evidence, without settlement authority.
110        acknowledgement: Box<IcMutationAcknowledgement>,
111    },
112}
113
114/// Admission failure after a returned snapshot upload reply.
115#[derive(Debug, Error)]
116pub enum IcSnapshotUploadReplyError {
117    /// Original selected journal cannot be re-admitted.
118    #[error(transparent)]
119    Journal(#[from] AttemptJournalError),
120    /// Passive exact reply association failed.
121    #[error(transparent)]
122    Association(#[from] IcSnapshotUploadAssociationError),
123    /// Original stage or ancestor evidence cannot be re-admitted.
124    #[error(transparent)]
125    Stage(#[from] ExecutionWorkflowPersistenceError),
126}
127
128#[cfg(all(test, unix))]
129mod tests;
ic_backup/workflow/ic_snapshot_upload/mod.rs

ic_backup/workflow/ic_snapshot_upload/
mod.rs
