ic_backup/workflow/ic_snapshot_upload/
mod.rs

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

ic_backup/workflow/ic_snapshot_upload/
mod.rs
