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 (acknowledgement, result) = super::ic_mutation::execute_reserved(
53        stage,
54        &authority,
55        |journal| {
56            original_authority(plan, operation_sequence, journal, payload)?;
57            Ok::<_, IcSnapshotUploadExecutionError<E>>(())
58        },
59        |journal| {
60            Ok(IcSnapshotUploadAttempt::new(
61                plan,
62                operation_sequence,
63                journal,
64                payload,
65            )?)
66        },
67        async |request| {
68            admit(request)
69                .await
70                .map_err(IcSnapshotUploadExecutionError::Admission)?;
71            stage.layout()?;
72            Ok(())
73        },
74        async |request, journal| Ok(provider.submit_upload(request, journal).await?),
75        async |journal, request, acknowledgement| {
76            let admitted = validate_acknowledgement(request, journal.record()?, acknowledgement)?;
77            stage.layout()?;
78            Ok::<_, IcSnapshotUploadReplyError>(complete(journal, request, &admitted).await)
79        },
80    )
81    .await?;
82    super::qualified_receipt::retain_reply(acknowledgement, result, |source, acknowledgement| {
83        IcSnapshotUploadExecutionError::AfterReply {
84            source,
85            acknowledgement,
86        }
87    })
88}
89
90/// Upload-step failures retain original spending and any bounded returned acknowledgement.
91#[derive(Debug, Error)]
92pub enum IcSnapshotUploadExecutionError<E: std::error::Error + 'static> {
93    /// Original operation authority cannot be derived.
94    #[error(transparent)]
95    Plan(#[from] OperationPlanError),
96    /// Workflow, stage or original ancestor evidence failed admission.
97    #[error(transparent)]
98    Stage(#[from] ExecutionWorkflowPersistenceError),
99    /// Original selected journal admission failed.
100    #[error(transparent)]
101    Journal(#[from] AttemptJournalError),
102    /// Complete original progress, prerequisites or durable reservation failed.
103    #[error(transparent)]
104    Progress(#[from] ExecutionProgressPersistenceError),
105    /// Canonical original mutation binding or current reservation differs.
106    #[error(transparent)]
107    Request(#[from] IcSnapshotUploadAttemptError),
108    /// Fresh integration-owned admission rejected after reservation.
109    #[error("fresh snapshot upload admission failed: {0}")]
110    Admission(#[source] E),
111    /// The single provider call failed; spending remains pending.
112    #[error(transparent)]
113    Provider(#[from] IcMutationProviderError),
114    /// Admission after a reply failed, retaining its complete bounded evidence.
115    #[error("snapshot upload reply rejected: {source}")]
116    AfterReply {
117        /// Typed original admission failure.
118        source: IcSnapshotUploadReplyError,
119        /// Complete returned passive evidence, without settlement authority.
120        acknowledgement: Box<IcMutationAcknowledgement>,
121    },
122}
123
124/// Admission failure after a returned snapshot upload reply.
125#[derive(Debug, Error)]
126pub enum IcSnapshotUploadReplyError {
127    /// Original selected journal cannot be re-admitted.
128    #[error(transparent)]
129    Journal(#[from] AttemptJournalError),
130    /// Passive exact reply association failed.
131    #[error(transparent)]
132    Association(#[from] IcSnapshotUploadAssociationError),
133    /// Original stage or ancestor evidence cannot be re-admitted.
134    #[error(transparent)]
135    Stage(#[from] ExecutionWorkflowPersistenceError),
136}
137
138#[cfg(all(test, unix))]
139mod tests;
ic_backup/workflow/ic_snapshot_upload/mod.rs

ic_backup/workflow/ic_snapshot_upload/
mod.rs
