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 admit_reply = || {
113        let admitted = validate_acknowledgement(&request, journal.record()?, &acknowledgement)?;
114        stage.layout()?;
115        Ok::<_, IcSnapshotUploadReplyError>(admitted)
116    };
117    let admitted = match admit_reply() {
118        Ok(admitted) => admitted,
119        Err(source) => {
120            return Err(IcSnapshotUploadExecutionError::AfterReply {
121                source,
122                acknowledgement: Box::new(acknowledgement),
123            });
124        }
125    };
126    let result = complete(&mut journal, &request, &admitted).await;
127    Ok((acknowledgement, result))
128}
129
130/// Upload-step failures retain original spending and any bounded returned acknowledgement.
131#[derive(Debug, Error)]
132pub enum IcSnapshotUploadExecutionError<E: std::error::Error + 'static> {
133    /// Original operation authority cannot be derived.
134    #[error(transparent)]
135    Plan(#[from] OperationPlanError),
136    /// Workflow, stage or original ancestor evidence failed admission.
137    #[error(transparent)]
138    Stage(#[from] ExecutionWorkflowPersistenceError),
139    /// Original selected journal admission failed.
140    #[error(transparent)]
141    Journal(#[from] AttemptJournalError),
142    /// Complete original progress, prerequisites or durable reservation failed.
143    #[error(transparent)]
144    Progress(#[from] ExecutionProgressPersistenceError),
145    /// Canonical original mutation binding or current reservation differs.
146    #[error(transparent)]
147    Request(#[from] IcSnapshotUploadAttemptError),
148    /// Fresh integration-owned admission rejected after reservation.
149    #[error("fresh snapshot upload admission failed: {0}")]
150    Admission(#[source] E),
151    /// The single provider call failed; spending remains pending.
152    #[error(transparent)]
153    Provider(#[from] IcMutationProviderError),
154    /// Admission after a reply failed, retaining its complete bounded evidence.
155    #[error("snapshot upload reply rejected: {source}")]
156    AfterReply {
157        /// Typed original admission failure.
158        source: IcSnapshotUploadReplyError,
159        /// Complete returned passive evidence, without settlement authority.
160        acknowledgement: Box<IcMutationAcknowledgement>,
161    },
162}
163
164/// Admission failure after a returned snapshot upload reply.
165#[derive(Debug, Error)]
166pub enum IcSnapshotUploadReplyError {
167    /// Original selected journal cannot be re-admitted.
168    #[error(transparent)]
169    Journal(#[from] AttemptJournalError),
170    /// Passive exact reply association failed.
171    #[error(transparent)]
172    Association(#[from] IcSnapshotUploadAssociationError),
173    /// Original stage or ancestor evidence cannot be re-admitted.
174    #[error(transparent)]
175    Stage(#[from] ExecutionWorkflowPersistenceError),
176}
177
178#[cfg(all(test, unix))]
179mod tests;
ic_backup/workflow/ic_snapshot_upload/mod.rs

ic_backup/workflow/ic_snapshot_upload/
mod.rs
