ic_backup/workflow/ic_snapshot_upload/data/
mod.rs

1//! Complete finite original data uploads with explicit integration-qualified receipts.
2
3use super::{IcSnapshotUploadExecutionError, upload_snapshot_with};
4use crate::model::data_stage_plan::SnapshotDataPlanningError;
5use crate::{
6    model::{
7        artifacts::ArtifactChecksumRecord,
8        attempt_journal::MutationReceiptRequest,
9        evidence_digest::EvidenceDigest,
10        ic_mutation::IcMutationAcknowledgement,
11        ic_snapshot_upload::{IcSnapshotDataUploadPlan, IcSnapshotUploadAttempt},
12    },
13    ops::persistence::{
14        AttemptJournalError, DownloadJournalGuard, ExecutionProgressPersistenceError,
15        ExecutionStageGuard, ExecutionWorkflowPersistenceError, IcSnapshotUploadArtifactError,
16        read_execution_progress,
17    },
18    ports::ic_snapshot_upload::IcSnapshotUploadProvider,
19};
20use thiserror::Error;
21
22/// Upload every exact original data operation and return qualified learned evidence.
23///
24/// Prepare the source-derived data plan, allocation predecessor, retained stage and
25/// complete original journals first; hold no attempt guards. Empty plans and any prior
26/// consumption reject before dispatch: this is a fresh stage, never partial upload
27/// resume. Original source and destination/command custody remain integration-owned.
28/// Fresh guarded preparation buffers one payload and rechecks its immutable binding
29/// before existing single-upload reservation, mandatory fresh `admit` and one call.
30///
31/// `qualify` must independently authenticate exact original data-write attribution and
32/// durably retain original request/reply bytes before returning its explicit Applied
33/// receipt under the selected journal lock. An empty reply never supplies a receipt.
34/// Account remote qualification separately before its calls. Record through the sole
35/// attempt owner before the next dependency can dispatch; failures stop immediately.
36/// Admission, submission and qualification are async under their selected journal
37/// guards. Retain each exact reply before awaiting cancellable qualification work.
38/// Cancellation preserves source bytes/references, earlier Applied receipts and
39/// the current pending reservation, denying whole-stage or single-call reentry.
40///
41/// After all exact receipts, return the full original plan and ordered canonical
42/// request/raw-reply evidence digest. The workflow owner publishes the checkpoint
43/// after journal exclusion ends; direct callers explicitly use that same stage owner.
44/// This is declared stage settlement, not authenticated whole-backend completeness,
45/// same-release load/start admission, terminal proof or fence/source-reference release.
46/// No hidden observation, reissue, refund, default provider or journal is added.
47/// # Errors
48/// Rejects changed originals, consumed attempts, source/preparation failures, spending,
49/// admission/provider/reply/qualification/receipt failure. Bounded returned
50/// replies survive post-reply rejection; spending, recorded receipts and source remain.
51pub async fn upload_snapshot_data<E: std::error::Error + 'static>(
52    stage: &ExecutionStageGuard<'_>,
53    upload: &IcSnapshotDataUploadPlan<'_, '_, '_>,
54    source: &DownloadJournalGuard<'_>,
55    snapshot: &str,
56    provider: &mut impl IcSnapshotUploadProvider,
57    mut admit: impl AsyncFnMut(&IcSnapshotUploadAttempt<'_, '_>) -> Result<(), E>,
58    mut qualify: impl AsyncFnMut(
59        &IcSnapshotUploadAttempt<'_, '_>,
60        &IcMutationAcknowledgement,
61    ) -> Result<MutationReceiptRequest, E>,
62) -> Result<ArtifactChecksumRecord, IcSnapshotDataUploadExecutionError<E>> {
63    upload.validate_binding(stage.binding())?;
64    let plan = upload
65        .plan()
66        .ok_or(SnapshotDataPlanningError::NoDataOperations)?;
67    if stage.plan() != plan {
68        return Err(SnapshotDataPlanningError::OriginalMismatch.into());
69    }
70    let intent = plan.digest();
71    let progress = read_execution_progress(stage.layout()?, &intent)?;
72    if !progress.is_unattempted() {
73        return Err(IcSnapshotDataUploadExecutionError::AlreadyAttempted);
74    }
75    let mut evidence = EvidenceDigest::new(b"ic-backup/ic-snapshot-data-upload/v1\0");
76    evidence.checksum(&intent);
77    let session = source.upload_session(snapshot, upload)?;
78    // Preparation binds each kind to one ordered operation in this same plan.
79    for (operation, kind) in plan.operations().iter().zip(upload.kinds()) {
80        let sequence = operation.operation_sequence();
81        let payload = session.prepare(sequence, kind.clone())?;
82        // The single-call owner rechecks the exact full source/context/request before spending.
83        let (acknowledgement, settlement) = upload_snapshot_with(
84            stage,
85            sequence,
86            &payload,
87            provider,
88            &mut admit,
89            async |journal, request, admitted| {
90                let receipt = qualify(request, admitted.acknowledgement())
91                    .await
92                    .map_err(IcSnapshotDataUploadReplyError::Qualification)?;
93                super::super::qualified_receipt::record_applied(
94                    journal,
95                    receipt,
96                    request.mutation_attempt(),
97                    payload.binding_digest().hash(),
98                    IcSnapshotDataUploadReplyError::ReceiptRequired,
99                    || {
100                        stage.layout()?;
101                        Ok(())
102                    },
103                )?;
104                stage.layout()?;
105                Ok::<_, IcSnapshotDataUploadReplyError<E>>(admitted.reply().digest())
106            },
107        )
108        .await?;
109        let digest =
110            settlement.map_err(|source| IcSnapshotDataUploadExecutionError::AfterReply {
111                operation_sequence: sequence,
112                source,
113                acknowledgement: Box::new(acknowledgement),
114            })?;
115        evidence.append(digest.hash().as_bytes());
116    }
117    session.finish()?;
118    let progress = read_execution_progress(stage.layout()?, &intent)?;
119    if !progress.all_applied() {
120        return Err(IcSnapshotDataUploadExecutionError::AlreadyAttempted);
121    }
122    Ok(evidence.finish())
123}
124
125/// Data-stage failure preserves source, original spending and returned acknowledgement bytes.
126#[derive(Debug, Error)]
127pub enum IcSnapshotDataUploadExecutionError<E: std::error::Error + 'static> {
128    /// Fresh-only entrypoint rejects any original consumed mutation or observation.
129    #[error("snapshot data upload original stage was already attempted")]
130    AlreadyAttempted,
131    /// Existing exact original data-plan/binding admission.
132    #[error(transparent)]
133    Planning(#[from] SnapshotDataPlanningError),
134    /// Existing retained workflow/stage/ancestor admission.
135    #[error(transparent)]
136    Stage(#[from] ExecutionWorkflowPersistenceError),
137    /// Complete original progress admission.
138    #[error(transparent)]
139    Progress(#[from] ExecutionProgressPersistenceError),
140    /// Fresh exact source preparation failed before the selected operation reserves.
141    #[error(transparent)]
142    Source(#[from] IcSnapshotUploadArtifactError),
143    /// One exact original upload failed; its owner retains bounded returned evidence.
144    #[error(transparent)]
145    Upload(#[from] IcSnapshotUploadExecutionError<E>),
146    /// Receipt settlement failed after a bounded returned acknowledgement.
147    #[error("snapshot data upload reply rejected for operation {operation_sequence}: {source}")]
148    AfterReply {
149        /// Exact original failed operation sequence.
150        operation_sequence: u64,
151        /// Original qualification or persistence rejection.
152        source: IcSnapshotDataUploadReplyError<E>,
153        /// Exact bounded reply, without an inferred outcome.
154        acknowledgement: Box<IcMutationAcknowledgement>,
155    },
156}
157
158/// Post-reply data-write rejection, without a second outcome/accounting owner.
159#[derive(Debug, Error)]
160pub enum IcSnapshotDataUploadReplyError<E: std::error::Error + 'static> {
161    /// Independent attribution/authentication or durable original-byte retention failed.
162    #[error("snapshot data upload qualification failed: {0}")]
163    Qualification(#[source] E),
164    /// Require the explicit exact original Applied receipt.
165    #[error("snapshot data upload requires an explicit original Applied receipt")]
166    ReceiptRequired,
167    /// Original workflow/stage/ancestor admission failed.
168    #[error(transparent)]
169    Stage(#[from] ExecutionWorkflowPersistenceError),
170    /// Existing selected journal/receipt persistence failed.
171    #[error(transparent)]
172    Journal(#[from] AttemptJournalError),
173}
174
175impl<E: std::error::Error + 'static> IcSnapshotDataUploadExecutionError<E> {
176    /// Borrow the bounded returned evidence carried by this error, including nested failures.
177    /// Absence describes retained evidence only and grants no dispatch or retry authority.
178    #[must_use]
179    pub fn acknowledgement(&self) -> Option<&IcMutationAcknowledgement> {
180        match self {
181            Self::AfterReply {
182                acknowledgement, ..
183            } => Some(acknowledgement),
184            Self::Upload(source) => source.acknowledgement(),
185            _ => None,
186        }
187    }
188}
ic_backup/workflow/ic_snapshot_upload/data/mod.rs

ic_backup/workflow/ic_snapshot_upload/data/
mod.rs
