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

ic_backup/workflow/ic_snapshot_upload/
mod.rs
