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

ic_backup/workflow/ic_snapshot_upload/
mod.rs
