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

ic_backup/workflow/ic_snapshot_upload/data/
mod.rs
