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::{
5    model::{
6        attempt_journal::MutationReceiptRequest,
7        evidence_digest::EvidenceDigest,
8        execution_workflow::ExecutionStagePredecessorRecord,
9        ic_mutation::IcMutationAcknowledgement,
10        ic_snapshot_upload::{
11            IcSnapshotDataUploadPlan, IcSnapshotDataUploadPlanningError, IcSnapshotUploadAttempt,
12        },
13    },
14    ops::persistence::{
15        AttemptJournalError, DownloadJournalGuard, ExecutionProgressPersistenceError,
16        ExecutionStageCheckpointError, ExecutionStageGuard, ExecutionWorkflowPersistenceError,
17        IcSnapshotUploadArtifactError, read_execution_progress,
18    },
19    ports::ic_snapshot_upload::IcSnapshotUploadProvider,
20};
21use thiserror::Error;
22
23/// Upload every exact original data operation, then checkpoint qualified reply evidence.
24///
25/// Prepare the source-derived data plan, allocation predecessor, retained stage and
26/// complete original journals first; hold no attempt guards. Empty plans and any prior
27/// consumption reject before dispatch: this is a fresh stage, never partial upload
28/// resume. Original source and destination/command custody remain integration-owned.
29/// Fresh guarded preparation buffers one payload and rechecks its immutable binding
30/// before existing single-upload reservation, mandatory fresh `admit` and one call.
31///
32/// `qualify` must independently authenticate exact original data-write attribution and
33/// durably retain original request/reply bytes before returning its explicit Applied
34/// receipt under the selected journal lock. An empty reply never supplies a receipt.
35/// Account remote qualification separately before its calls. Record through the sole
36/// attempt owner before the next dependency can dispatch; failures stop immediately.
37/// Admission, submission and qualification are async under their selected journal
38/// guards. Retain each exact reply before awaiting cancellable qualification work.
39/// Cancellation preserves source bytes/references, earlier Applied receipts and
40/// the current pending reservation, denying whole-stage or single-call reentry.
41///
42/// After all exact receipts, release journal locks and checkpoint the full original plan
43/// and ordered canonical request/raw-reply evidence through the existing 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 or checkpoint 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<ExecutionStagePredecessorRecord, IcSnapshotDataUploadExecutionError<E>> {
63    upload.validate_binding(stage.binding())?;
64    let plan = upload
65        .plan()
66        .ok_or(IcSnapshotDataUploadPlanningError::NoDataWrites)?;
67    if stage.plan() != plan {
68        return Err(IcSnapshotDataUploadPlanningError::OriginalMismatch.into());
69    }
70    let progress = read_execution_progress(stage.layout()?, &plan.digest())?;
71    if progress.attempts.mutations_used != 0 || progress.attempts.observations_used != 0 {
72        return Err(IcSnapshotDataUploadExecutionError::AlreadyAttempted);
73    }
74    let mut evidence = EvidenceDigest::new(b"ic-backup/ic-snapshot-data-upload/v1\0");
75    evidence.append(plan.digest().hash().as_bytes());
76    for (index, kind) in upload.kinds().iter().enumerate() {
77        let sequence =
78            u64::try_from(index).map_err(|_| IcSnapshotDataUploadPlanningError::CountOverflow)?;
79        let metadata = upload.metadata();
80        let payload = source.prepare_ic_snapshot_upload_data(
81            metadata.source_plan(),
82            snapshot,
83            metadata,
84            upload.destination(),
85            kind.clone(),
86        )?;
87        // The single-call owner rechecks the exact full source/context/request before spending.
88        let (acknowledgement, settlement) = upload_snapshot_with(
89            stage,
90            sequence,
91            &payload,
92            provider,
93            &mut admit,
94            async |journal, request, admitted| {
95                super::super::qualified_receipt::AppliedReceiptRecorder {
96                    journal,
97                    expected_attempt: request.mutation_attempt(),
98                    expected_request: payload.binding_digest().hash(),
99                }
100                .record_applied(
101                    request,
102                    admitted.acknowledgement(),
103                    async |request, reply| qualify(request, reply).await,
104                    IcSnapshotDataUploadReplyError::Qualification,
105                    || IcSnapshotDataUploadReplyError::ReceiptRequired,
106                    || {
107                        stage
108                            .layout()
109                            .map(|_| ())
110                            .map_err(IcSnapshotDataUploadReplyError::Stage)
111                    },
112                )
113                .await?;
114                stage.layout()?;
115                Ok::<_, IcSnapshotDataUploadReplyError<E>>(admitted.reply().digest())
116            },
117        )
118        .await?;
119        let digest =
120            settlement.map_err(|source| IcSnapshotDataUploadExecutionError::AfterReply {
121                operation_sequence: sequence,
122                source,
123                acknowledgement: Box::new(acknowledgement),
124            })?;
125        evidence.append(digest.hash().as_bytes());
126    }
127    let progress = read_execution_progress(stage.layout()?, &plan.digest())?;
128    if progress.applied_operations != upload.kinds().len() {
129        return Err(IcSnapshotDataUploadExecutionError::AlreadyAttempted);
130    }
131    Ok(stage.reconcile_checkpoint(evidence.finish())?)
132}
133
134/// Data-stage failure preserves source, original spending and returned acknowledgement bytes.
135#[derive(Debug, Error)]
136pub enum IcSnapshotDataUploadExecutionError<E: std::error::Error + 'static> {
137    /// Fresh-only entrypoint rejects any original consumed mutation or observation.
138    #[error("snapshot data upload original stage was already attempted")]
139    AlreadyAttempted,
140    /// Existing exact original data-plan/binding admission.
141    #[error(transparent)]
142    Planning(#[from] IcSnapshotDataUploadPlanningError),
143    /// Existing retained workflow/stage/ancestor admission.
144    #[error(transparent)]
145    Stage(#[from] ExecutionWorkflowPersistenceError),
146    /// Complete original progress admission.
147    #[error(transparent)]
148    Progress(#[from] ExecutionProgressPersistenceError),
149    /// Fresh exact source preparation failed before the selected operation reserves.
150    #[error(transparent)]
151    Source(#[from] IcSnapshotUploadArtifactError),
152    /// One exact original upload failed; its owner retains bounded returned evidence.
153    #[error(transparent)]
154    Upload(#[from] IcSnapshotUploadExecutionError<E>),
155    /// Receipt settlement failed after a bounded returned acknowledgement.
156    #[error("snapshot data upload reply rejected for operation {operation_sequence}: {source}")]
157    AfterReply {
158        /// Exact original failed operation sequence.
159        operation_sequence: u64,
160        /// Original qualification or persistence rejection.
161        source: IcSnapshotDataUploadReplyError<E>,
162        /// Exact bounded reply, without an inferred outcome.
163        acknowledgement: Box<IcMutationAcknowledgement>,
164    },
165    /// Existing all-Applied stage checkpoint failed; receipts and occupied evidence remain.
166    #[error(transparent)]
167    Checkpoint(#[from] ExecutionStageCheckpointError),
168}
169
170/// Post-reply data-write rejection, without a second outcome/accounting owner.
171#[derive(Debug, Error)]
172pub enum IcSnapshotDataUploadReplyError<E: std::error::Error + 'static> {
173    /// Independent attribution/authentication or durable original-byte retention failed.
174    #[error("snapshot data upload qualification failed: {0}")]
175    Qualification(#[source] E),
176    /// Require the explicit exact original Applied receipt.
177    #[error("snapshot data upload requires an explicit original Applied receipt")]
178    ReceiptRequired,
179    /// Original workflow/stage/ancestor admission failed.
180    #[error(transparent)]
181    Stage(#[from] ExecutionWorkflowPersistenceError),
182    /// Existing selected journal/receipt persistence failed.
183    #[error(transparent)]
184    Journal(#[from] AttemptJournalError),
185}
ic_backup/workflow/ic_snapshot_upload/data/mod.rs

ic_backup/workflow/ic_snapshot_upload/data/
mod.rs
