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

ic_backup/workflow/ic_snapshot_upload/data/
mod.rs
