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/// Durably reserve one exact original upload, freshly admit it and submit once.
37///
38/// Prepare the retained stage and every original journal before entry; hold no attempt
39/// guards. Guarded source preparation and exact metadata/tree/payload retention precede
40/// the upload plan. Data bytes and their independently attributed destination must be
41/// bound before that data stage is published, within its original workflow allocation.
42/// This checks the canonical full source/upload context and binding before spending,
43/// then delegates reservation/prerequisites to the existing complete-plan owner.
44/// Missing, pending, Applied or exhausted originals never reach provider invocation.
45///
46/// The mandatory fallible `admit` callback qualifies authentic complete source bytes,
47/// actual fresh controllers and application/restore requirements, exact metadata/data
48/// destination attribution, stable source/payload custody and exclusive never-dispatched
49/// command custody. It runs under the selected journal lock; remote preflight requires
50/// its own prior accounting. Local checks grant no byte fence or allocation authority.
51/// The existing provider sends only the exact accounted replicated update, with no
52/// batching, retries, hidden observations or implicit funding. No default is installed.
53/// Await admission and submission under the selected original guard. Cancellation
54/// releases exclusion without changing spending or permitting reentry. The provider
55/// receives the currently guarded original journal; the core supplies no executor.
56///
57/// Re-admit original stage/ancestor evidence before and after dispatch and check the
58/// bounded acknowledgement through the canonical upload policy/decoder. Success
59/// remains pending; integrations independently authenticate attribution and retain
60/// exact replies before explicitly recording receipts with the existing journal owner.
61/// A returned ID/empty reply alone grants no allocation outcome, learned data authority,
62/// complete transfer, load/start admission, terminal or fence/source-reference release.
63/// # Errors
64/// Rejects changed originals/payload/context, spending, fresh admission, provider or
65/// reply failures. Post-reservation failures retain consumption; post-reply rejections
66/// retain the bounded acknowledgement. No failure grants redispatch/refund/cleanup.
67pub async fn upload_snapshot<E: std::error::Error + 'static>(
68    stage: &ExecutionStageGuard<'_>,
69    operation_sequence: u64,
70    payload: &IcSnapshotUploadRequest<'_>,
71    provider: &mut impl IcSnapshotUploadProvider,
72    admit: impl AsyncFnOnce(&IcSnapshotUploadAttempt<'_, '_>) -> Result<(), E>,
73) -> Result<IcMutationAcknowledgement, IcSnapshotUploadExecutionError<E>> {
74    let (acknowledgement, ()) = upload_snapshot_with(
75        stage,
76        operation_sequence,
77        payload,
78        provider,
79        admit,
80        async |_, _, _| (),
81    )
82    .await?;
83    Ok(acknowledgement)
84}
85
86// Complete compound uploads under their original guard using the admitted reply.
87// Return before checkpointing, which must run after journal exclusion is released.
88async fn upload_snapshot_with<E: std::error::Error + 'static, T>(
89    stage: &ExecutionStageGuard<'_>,
90    operation_sequence: u64,
91    payload: &IcSnapshotUploadRequest<'_>,
92    provider: &mut impl IcSnapshotUploadProvider,
93    admit: impl AsyncFnOnce(&IcSnapshotUploadAttempt<'_, '_>) -> Result<(), E>,
94    complete: impl AsyncFnOnce(
95        &mut AttemptJournalGuard<'_>,
96        &IcSnapshotUploadAttempt<'_, '_>,
97        &IcSnapshotUploadAcknowledgementView<'_, '_>,
98    ) -> T,
99) -> Result<(IcMutationAcknowledgement, T), IcSnapshotUploadExecutionError<E>> {
100    let plan = stage.plan();
101    let authority = plan.attempt_authority(operation_sequence)?;
102    let mut journal = AttemptJournalGuard::open(stage.layout()?, &authority)?;
103    original_authority(plan, operation_sequence, journal.record()?, payload)?;
104    journal.reserve_planned_mutation(&plan.digest())?;
105    let request =
106        IcSnapshotUploadAttempt::new(plan, operation_sequence, journal.record()?, payload)?;
107    admit(&request)
108        .await
109        .map_err(IcSnapshotUploadExecutionError::Admission)?;
110    stage.layout()?;
111    let acknowledgement = provider.submit_upload(&request, journal.record()?).await?;
112    let record = match journal.record() {
113        Ok(record) => record,
114        Err(source) => {
115            return Err(IcSnapshotUploadExecutionError::AfterReplyJournal {
116                source,
117                acknowledgement: Box::new(acknowledgement),
118            });
119        }
120    };
121    let admitted = match validate_acknowledgement(&request, record, &acknowledgement) {
122        Ok(admitted) => admitted,
123        Err(source) => {
124            return Err(IcSnapshotUploadExecutionError::Association {
125                source,
126                acknowledgement: Box::new(acknowledgement),
127            });
128        }
129    };
130    if let Err(source) = stage.layout() {
131        return Err(IcSnapshotUploadExecutionError::AfterReplyStage {
132            source,
133            acknowledgement: Box::new(acknowledgement),
134        });
135    }
136    let result = complete(&mut journal, &request, &admitted).await;
137    Ok((acknowledgement, result))
138}
139
140/// Upload-step failures retain original spending and any bounded returned acknowledgement.
141#[derive(Debug, Error)]
142pub enum IcSnapshotUploadExecutionError<E: std::error::Error + 'static> {
143    /// Original operation authority cannot be derived.
144    #[error(transparent)]
145    Plan(#[from] OperationPlanError),
146    /// Workflow, stage or original ancestor evidence failed admission.
147    #[error(transparent)]
148    Stage(#[from] ExecutionWorkflowPersistenceError),
149    /// Original selected journal admission failed.
150    #[error(transparent)]
151    Journal(#[from] AttemptJournalError),
152    /// Complete original progress, prerequisites or durable reservation failed.
153    #[error(transparent)]
154    Progress(#[from] ExecutionProgressPersistenceError),
155    /// Canonical original mutation binding or current reservation differs.
156    #[error(transparent)]
157    Request(#[from] IcSnapshotUploadAttemptError),
158    /// Fresh integration-owned admission rejected after reservation.
159    #[error("fresh snapshot upload admission failed: {0}")]
160    Admission(#[source] E),
161    /// The single provider call failed; spending remains pending.
162    #[error(transparent)]
163    Provider(#[from] IcMutationProviderError),
164    /// Passive bounded association rejected a returned acknowledgement.
165    #[error("snapshot upload acknowledgement association failed: {source}")]
166    Association {
167        /// Existing canonical mutation association rejection.
168        source: IcSnapshotUploadAssociationError,
169        /// Exact returned acknowledgement, without authenticated outcome.
170        acknowledgement: Box<IcMutationAcknowledgement>,
171    },
172    /// Selected journal re-admission failed after a reply.
173    #[error("snapshot upload journal changed after reply: {source}")]
174    AfterReplyJournal {
175        /// Original journal rejection.
176        source: AttemptJournalError,
177        /// Exact bounded returned acknowledgement.
178        acknowledgement: Box<IcMutationAcknowledgement>,
179    },
180    /// Stage or ancestor re-admission failed after a reply.
181    #[error("snapshot upload stage changed after reply: {source}")]
182    AfterReplyStage {
183        /// Original stage or ancestor rejection.
184        source: ExecutionWorkflowPersistenceError,
185        /// Exact bounded returned acknowledgement.
186        acknowledgement: Box<IcMutationAcknowledgement>,
187    },
188}
189
190#[cfg(all(test, unix))]
191mod tests;
ic_backup/workflow/ic_snapshot_upload/mod.rs

ic_backup/workflow/ic_snapshot_upload/
mod.rs
