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        attempt_journal::MutationReceiptRequest,
8        evidence_digest::EvidenceDigest,
9        execution_workflow::ExecutionStagePredecessorRecord,
10        ic_mutation::IcMutationAcknowledgement,
11        ic_snapshot_upload::{IcSnapshotDataUploadPlan, IcSnapshotUploadAttempt},
12    },
13    ops::persistence::{
14        AttemptJournalError, DownloadJournalGuard, ExecutionProgressPersistenceError,
15        ExecutionStageCheckpointError, ExecutionStageGuard, ExecutionWorkflowPersistenceError,
16        IcSnapshotUploadArtifactError, read_execution_progress,
17    },
18    ports::ic_snapshot_upload::IcSnapshotUploadProvider,
19};
20use thiserror::Error;
21
22/// Upload every exact original data operation, then checkpoint qualified reply 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, release journal locks and checkpoint the full original plan
42/// and ordered canonical request/raw-reply evidence through the existing stage owner.
43/// This is declared stage settlement, not authenticated whole-backend completeness,
44/// same-release load/start admission, terminal proof or fence/source-reference release.
45/// No hidden observation, reissue, refund, default provider or journal is added.
46/// # Errors
47/// Rejects changed originals, consumed attempts, source/preparation failures, spending,
48/// admission/provider/reply/qualification/receipt or checkpoint failure. Bounded returned
49/// replies survive post-reply rejection; spending, recorded receipts and source remain.
50pub async fn upload_snapshot_data<E: std::error::Error + 'static>(
51    stage: &ExecutionStageGuard<'_>,
52    upload: &IcSnapshotDataUploadPlan<'_, '_, '_>,
53    source: &DownloadJournalGuard<'_>,
54    snapshot: &str,
55    provider: &mut impl IcSnapshotUploadProvider,
56    mut admit: impl AsyncFnMut(&IcSnapshotUploadAttempt<'_, '_>) -> Result<(), E>,
57    mut qualify: impl AsyncFnMut(
58        &IcSnapshotUploadAttempt<'_, '_>,
59        &IcMutationAcknowledgement,
60    ) -> Result<MutationReceiptRequest, E>,
61) -> Result<ExecutionStagePredecessorRecord, IcSnapshotDataUploadExecutionError<E>> {
62    upload.validate_binding(stage.binding())?;
63    let plan = upload
64        .plan()
65        .ok_or(SnapshotDataPlanningError::NoDataOperations)?;
66    if stage.plan() != plan {
67        return Err(SnapshotDataPlanningError::OriginalMismatch.into());
68    }
69    let progress = read_execution_progress(stage.layout()?, &plan.digest())?;
70    if !progress.is_unattempted() {
71        return Err(IcSnapshotDataUploadExecutionError::AlreadyAttempted);
72    }
73    let mut evidence = EvidenceDigest::new(b"ic-backup/ic-snapshot-data-upload/v1\0");
74    evidence.append(plan.digest().hash().as_bytes());
75    let session = source.upload_session(snapshot, upload)?;
76    // Preparation binds each kind to one ordered operation in this same plan.
77    for (operation, kind) in plan.operations().iter().zip(upload.kinds()) {
78        let sequence = operation.operation_sequence();
79        let payload = session.prepare(sequence, kind.clone())?;
80        // The single-call owner rechecks the exact full source/context/request before spending.
81        let (acknowledgement, settlement) = upload_snapshot_with(
82            stage,
83            sequence,
84            &payload,
85            provider,
86            &mut admit,
87            async |journal, request, admitted| {
88                let receipt = qualify(request, admitted.acknowledgement())
89                    .await
90                    .map_err(IcSnapshotDataUploadReplyError::Qualification)?;
91                super::super::qualified_receipt::record_applied(
92                    journal,
93                    receipt,
94                    request.mutation_attempt(),
95                    payload.binding_digest().hash(),
96                    IcSnapshotDataUploadReplyError::ReceiptRequired,
97                    || {
98                        stage.layout()?;
99                        Ok(())
100                    },
101                )?;
102                stage.layout()?;
103                Ok::<_, IcSnapshotDataUploadReplyError<E>>(admitted.reply().digest())
104            },
105        )
106        .await?;
107        let digest =
108            settlement.map_err(|source| IcSnapshotDataUploadExecutionError::AfterReply {
109                operation_sequence: sequence,
110                source,
111                acknowledgement: Box::new(acknowledgement),
112            })?;
113        evidence.append(digest.hash().as_bytes());
114    }
115    session.finish()?;
116    let progress = read_execution_progress(stage.layout()?, &plan.digest())?;
117    if !progress.all_applied() {
118        return Err(IcSnapshotDataUploadExecutionError::AlreadyAttempted);
119    }
120    Ok(stage.reconcile_checkpoint(evidence.finish())?)
121}
122
123/// Data-stage failure preserves source, original spending and returned acknowledgement bytes.
124#[derive(Debug, Error)]
125pub enum IcSnapshotDataUploadExecutionError<E: std::error::Error + 'static> {
126    /// Fresh-only entrypoint rejects any original consumed mutation or observation.
127    #[error("snapshot data upload original stage was already attempted")]
128    AlreadyAttempted,
129    /// Existing exact original data-plan/binding admission.
130    #[error(transparent)]
131    Planning(#[from] SnapshotDataPlanningError),
132    /// Existing retained workflow/stage/ancestor admission.
133    #[error(transparent)]
134    Stage(#[from] ExecutionWorkflowPersistenceError),
135    /// Complete original progress admission.
136    #[error(transparent)]
137    Progress(#[from] ExecutionProgressPersistenceError),
138    /// Fresh exact source preparation failed before the selected operation reserves.
139    #[error(transparent)]
140    Source(#[from] IcSnapshotUploadArtifactError),
141    /// One exact original upload failed; its owner retains bounded returned evidence.
142    #[error(transparent)]
143    Upload(#[from] IcSnapshotUploadExecutionError<E>),
144    /// Receipt settlement failed after a bounded returned acknowledgement.
145    #[error("snapshot data upload reply rejected for operation {operation_sequence}: {source}")]
146    AfterReply {
147        /// Exact original failed operation sequence.
148        operation_sequence: u64,
149        /// Original qualification or persistence rejection.
150        source: IcSnapshotDataUploadReplyError<E>,
151        /// Exact bounded reply, without an inferred outcome.
152        acknowledgement: Box<IcMutationAcknowledgement>,
153    },
154    /// Existing all-Applied stage checkpoint failed; receipts and occupied evidence remain.
155    #[error(transparent)]
156    Checkpoint(#[from] ExecutionStageCheckpointError),
157}
158
159/// Post-reply data-write rejection, without a second outcome/accounting owner.
160#[derive(Debug, Error)]
161pub enum IcSnapshotDataUploadReplyError<E: std::error::Error + 'static> {
162    /// Independent attribution/authentication or durable original-byte retention failed.
163    #[error("snapshot data upload qualification failed: {0}")]
164    Qualification(#[source] E),
165    /// Require the explicit exact original Applied receipt.
166    #[error("snapshot data upload requires an explicit original Applied receipt")]
167    ReceiptRequired,
168    /// Original workflow/stage/ancestor admission failed.
169    #[error(transparent)]
170    Stage(#[from] ExecutionWorkflowPersistenceError),
171    /// Existing selected journal/receipt persistence failed.
172    #[error(transparent)]
173    Journal(#[from] AttemptJournalError),
174}
ic_backup/workflow/ic_snapshot_upload/data/mod.rs

ic_backup/workflow/ic_snapshot_upload/data/
mod.rs
