ic_backup/workflow/ic_snapshot_download/
mod.rs

1//! Complete original planned data transfer with explicit integration-qualified receipts.
2
3use crate::{
4    model::{
5        artifacts::ArtifactChecksumRecord,
6        attempt_journal::MutationReceiptRequest,
7        ic_snapshot_download::{IcSnapshotDownloadPlan, IcSnapshotDownloadPlanningError},
8        ic_snapshot_transfer_read::{
9            IcSnapshotTransferReadError, IcSnapshotTransferReadPayload,
10            IcSnapshotTransferReadRequest, IcSnapshotTransferReadResponse,
11        },
12    },
13    ops::persistence::{
14        AttemptJournalError, AttemptJournalGuard, ExecutionProgressPersistenceError,
15        ExecutionStageGuard, ExecutionWorkflowPersistenceError, IcSnapshotArtifactError,
16        IcSnapshotArtifactWriter, read_execution_progress,
17    },
18    policy::ic_snapshot_transfer_read::{
19        IcSnapshotTransferReadAssociationError, IcSnapshotTransferReadReply,
20        IcSnapshotTransferReadView,
21    },
22    ports::ic_snapshot_transfer_read::IcSnapshotTransferReadProvider,
23    workflow::ic_snapshot_transfer_read::{
24        IcSnapshotTransferReadExecutionError, read_snapshot_with,
25    },
26};
27use thiserror::Error;
28
29/// Stream every original data request into an existing private writer and publish it.
30///
31/// The metadata read and its independently qualified receipt/checkpoint must already
32/// exist. Prepare the metadata-derived data stage and its complete original journals,
33/// retain exact token/raw-ID association, then create the writer with exact raw metadata.
34/// Hold no attempt guards on entry. Read-free plans and any consumed original attempt
35/// reject; this is a fresh transfer, never partial-read resume or artifact recovery.
36///
37/// For every request, `admit` qualifies fresh access, metadata/command custody and
38/// application requirements under the existing single-read coordinator. `qualify`
39/// must independently authenticate exact original response attribution and return an
40/// explicit original Applied receipt. Wire shape alone cannot qualify that receipt.
41/// It runs under the same original journal lock held through dispatch and reuses
42/// that read's decoded reply. Append exact decoded bytes before
43/// the sole journal owner records that receipt; only then may the next dependent read
44/// run. Failed qualification, append or persistence stops without another provider call.
45/// Admission, submission and qualification are awaited under their selected journal
46/// guards. Retain each reply durably before cancellable qualification work. Dropping
47/// this future consumes the writer, retaining partial bytes, earlier receipts and
48/// the current pending reservation; it grants no resume or repeat-call authority.
49///
50/// All original limits/records remain unchanged. Errors consume the writer and preserve
51/// partial bytes, pending spending and every recorded receipt. Returned bounded replies
52/// survive post-read rejection. Success uses the existing fresh checksum/durable
53/// publisher, but supplies no immutable manifest/checkpoint, complete product backup,
54/// terminal proof, default provider, refund or fence/reference release. Stable bytes,
55/// complete authenticated backend transfer and application safety stay integration-owned.
56/// # Errors
57/// Rejects changed stage/plan/metadata/writer, prior consumption, spending, admission,
58/// provider or receipt failures, malformed replies and local IO/publication failure.
59pub async fn download_snapshot<E: std::error::Error + 'static>(
60    stage: &ExecutionStageGuard<'_>,
61    download: &IcSnapshotDownloadPlan<'_, '_>,
62    mut writer: IcSnapshotArtifactWriter<'_, '_, '_>,
63    provider: &mut impl IcSnapshotTransferReadProvider,
64    mut admit: impl AsyncFnMut(&IcSnapshotTransferReadRequest<'_, '_>) -> Result<(), E>,
65    mut qualify: impl AsyncFnMut(
66        &IcSnapshotTransferReadRequest<'_, '_>,
67        &IcSnapshotTransferReadResponse,
68    ) -> Result<MutationReceiptRequest, E>,
69) -> Result<ArtifactChecksumRecord, IcSnapshotDownloadExecutionError<E>> {
70    download.validate_binding(stage.binding())?;
71    let plan = download
72        .plan()
73        .ok_or(IcSnapshotDownloadPlanningError::NoDataReads)?;
74    if stage.plan() != plan
75        || download
76            .requests()
77            .iter()
78            .any(|request| request.metadata().digest() != writer.coverage().metadata().digest())
79        || writer.coverage().covered_region_bytes() != [0; 3]
80        || writer.coverage().covered_chunks() != 0
81    {
82        return Err(IcSnapshotDownloadExecutionError::OriginalMismatch);
83    }
84    writer.validate_transfer_origin(stage.layout()?.root(), plan.digest().hash())?;
85    let progress = read_execution_progress(stage.layout()?, &plan.digest())?;
86    if progress.attempts.mutations_used != 0 || progress.attempts.observations_used != 0 {
87        return Err(IcSnapshotDownloadExecutionError::AlreadyAttempted);
88    }
89    for (index, payload) in download.requests().iter().enumerate() {
90        // Planning already bounded the ordinal by the original stage allowance.
91        let sequence =
92            u64::try_from(index).map_err(|_| IcSnapshotDownloadPlanningError::CountOverflow)?;
93        let (response, result) = read_snapshot_with(
94            stage,
95            sequence,
96            IcSnapshotTransferReadPayload::Data(payload),
97            provider,
98            &mut admit,
99            async |journal, request, admitted| {
100                append_qualified(stage, journal, request, admitted, writer, &mut qualify).await
101            },
102        )
103        .await?;
104        writer = result.map_err(|source| IcSnapshotDownloadExecutionError::AfterReply {
105            operation_sequence: sequence,
106            source,
107            response: Box::new(response),
108        })?;
109    }
110    let progress = read_execution_progress(stage.layout()?, &plan.digest())?;
111    if progress.applied_operations != download.requests().len() {
112        return Err(IcSnapshotDownloadExecutionError::AlreadyAttempted);
113    }
114    writer.validate_transfer_origin(stage.layout()?.root(), plan.digest().hash())?;
115    let checksum = writer.finish()?;
116    if let Err(source) = stage.layout() {
117        return Err(IcSnapshotDownloadExecutionError::AfterPublication { source, checksum });
118    }
119    Ok(checksum)
120}
121
122async fn append_qualified<'journal, 'layout, 'metadata, E: std::error::Error + 'static>(
123    stage: &ExecutionStageGuard<'_>,
124    journal: &mut AttemptJournalGuard<'_>,
125    request: &IcSnapshotTransferReadRequest<'_, '_>,
126    admitted: &IcSnapshotTransferReadView<'_, '_>,
127    writer: IcSnapshotArtifactWriter<'journal, 'layout, 'metadata>,
128    qualify: &mut impl AsyncFnMut(
129        &IcSnapshotTransferReadRequest<'_, '_>,
130        &IcSnapshotTransferReadResponse,
131    ) -> Result<MutationReceiptRequest, E>,
132) -> Result<IcSnapshotArtifactWriter<'journal, 'layout, 'metadata>, IcSnapshotDownloadReplyError<E>>
133{
134    let plan = stage.plan();
135    let writer = super::qualified_receipt::AppliedReceiptRecorder {
136        journal,
137        expected_attempt: request.mutation_attempt(),
138        expected_request: request.payload().digest().hash(),
139    }
140    .record_applied(
141        request,
142        admitted.response(),
143        async |request, reply| qualify(request, reply).await,
144        IcSnapshotDownloadReplyError::Qualification,
145        || IcSnapshotDownloadReplyError::ReceiptRequired,
146        || {
147            writer.validate_transfer_origin(stage.layout()?.root(), plan.digest().hash())?;
148            let IcSnapshotTransferReadReply::Data(reply) = admitted.reply() else {
149                return Err(IcSnapshotDownloadReplyError::ReceiptRequired);
150            };
151            Ok(writer.append(reply)?)
152        },
153    )
154    .await?;
155    stage.layout()?;
156    Ok(writer)
157}
158
159/// Failures preserve original spending and partial/published artifact evidence.
160#[derive(Debug, Error)]
161pub enum IcSnapshotDownloadExecutionError<E: std::error::Error + 'static> {
162    /// Exact retained writer, stage or metadata differs.
163    #[error("snapshot download originals differ")]
164    OriginalMismatch,
165    /// A prior attempt cannot be replayed by this fresh transfer entrypoint.
166    #[error("snapshot download original data stage was already attempted")]
167    AlreadyAttempted,
168    /// Existing original download plan/binding admission.
169    #[error(transparent)]
170    Planning(#[from] IcSnapshotDownloadPlanningError),
171    /// Existing retained stage/ancestor admission.
172    #[error(transparent)]
173    Stage(#[from] ExecutionWorkflowPersistenceError),
174    /// Complete original progress admission.
175    #[error(transparent)]
176    Progress(#[from] ExecutionProgressPersistenceError),
177    /// One original read failed; its owner retains any returned bounded reply.
178    #[error(transparent)]
179    Read(#[from] IcSnapshotTransferReadExecutionError<E>),
180    /// Canonical writer/checksum/durable publication failure.
181    #[error(transparent)]
182    Artifact(#[from] IcSnapshotArtifactError),
183    /// Rejection after a returned read retains the exact bounded response.
184    #[error("snapshot download reply rejected for operation {operation_sequence}: {source}")]
185    AfterReply {
186        /// Exact original data operation.
187        operation_sequence: u64,
188        /// Original rejection without undoing receipt or spending.
189        source: IcSnapshotDownloadReplyError<E>,
190        /// Exact bounded returned bytes and passive claims.
191        response: Box<IcSnapshotTransferReadResponse>,
192    },
193    /// Publication succeeded but stage re-admission failed; retain the published checksum.
194    #[error("snapshot download stage changed after publication: {source}")]
195    AfterPublication {
196        /// Retained stage/ancestor admission failure.
197        source: ExecutionWorkflowPersistenceError,
198        /// Existing durable publisher's returned checksum.
199        checksum: ArtifactChecksumRecord,
200    },
201}
202
203/// Rejection of a returned data reply while retaining its original attempt.
204#[derive(Debug, Error)]
205pub enum IcSnapshotDownloadReplyError<E: std::error::Error + 'static> {
206    /// Fresh authenticated exact attribution was not qualified by the integration.
207    #[error("snapshot download reply qualification failed: {0}")]
208    Qualification(#[source] E),
209    /// Require the explicit exact original Applied receipt, without outcome inference.
210    #[error("snapshot download requires an explicit original Applied receipt")]
211    ReceiptRequired,
212    /// Existing original stage admission.
213    #[error(transparent)]
214    Stage(#[from] ExecutionWorkflowPersistenceError),
215    /// Original selected journal/spending/receipt admission.
216    #[error(transparent)]
217    Journal(#[from] AttemptJournalError),
218    /// Original request/current reservation admission.
219    #[error(transparent)]
220    Request(#[from] IcSnapshotTransferReadError),
221    /// Existing bounded passive decoder/association.
222    #[error(transparent)]
223    Association(#[from] IcSnapshotTransferReadAssociationError),
224    /// Exact local bytes/custody admission failed.
225    #[error(transparent)]
226    Artifact(#[from] IcSnapshotArtifactError),
227}
228
229#[cfg(test)]
230mod tests;
ic_backup/workflow/ic_snapshot_download/mod.rs

ic_backup/workflow/ic_snapshot_download/
mod.rs
