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

ic_backup/workflow/ic_snapshot_download/
mod.rs
