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 receipt = qualify(request, admitted.response())
133        .await
134        .map_err(IcSnapshotDownloadReplyError::Qualification)?;
135    let receipt = super::qualified_receipt::admit_applied(
136        receipt,
137        request.mutation_attempt(),
138        request.payload().digest().hash(),
139    )
140    .ok_or(IcSnapshotDownloadReplyError::ReceiptRequired)?;
141    writer.validate_transfer_origin(stage.layout()?.root(), plan.digest().hash())?;
142    let IcSnapshotTransferReadReply::Data(reply) = admitted.reply() else {
143        return Err(IcSnapshotDownloadReplyError::ReceiptRequired);
144    };
145    let writer = writer.append(reply)?;
146    journal.record_mutation(receipt)?;
147    stage.layout()?;
148    Ok(writer)
149}
150
151/// Failures preserve original spending and partial/published artifact evidence.
152#[derive(Debug, Error)]
153pub enum IcSnapshotDownloadExecutionError<E: std::error::Error + 'static> {
154    /// Exact retained writer, stage or metadata differs.
155    #[error("snapshot download originals differ")]
156    OriginalMismatch,
157    /// A prior attempt cannot be replayed by this fresh transfer entrypoint.
158    #[error("snapshot download original data stage was already attempted")]
159    AlreadyAttempted,
160    /// Existing original download plan/binding admission.
161    #[error(transparent)]
162    Planning(#[from] IcSnapshotDownloadPlanningError),
163    /// Existing retained stage/ancestor admission.
164    #[error(transparent)]
165    Stage(#[from] ExecutionWorkflowPersistenceError),
166    /// Complete original progress admission.
167    #[error(transparent)]
168    Progress(#[from] ExecutionProgressPersistenceError),
169    /// One original read failed; its owner retains any returned bounded reply.
170    #[error(transparent)]
171    Read(#[from] IcSnapshotTransferReadExecutionError<E>),
172    /// Canonical writer/checksum/durable publication failure.
173    #[error(transparent)]
174    Artifact(#[from] IcSnapshotArtifactError),
175    /// Rejection after a returned read retains the exact bounded response.
176    #[error("snapshot download reply rejected for operation {operation_sequence}: {source}")]
177    AfterReply {
178        /// Exact original data operation.
179        operation_sequence: u64,
180        /// Original rejection without undoing receipt or spending.
181        source: IcSnapshotDownloadReplyError<E>,
182        /// Exact bounded returned bytes and passive claims.
183        response: Box<IcSnapshotTransferReadResponse>,
184    },
185    /// Publication succeeded but stage re-admission failed; retain the published checksum.
186    #[error("snapshot download stage changed after publication: {source}")]
187    AfterPublication {
188        /// Retained stage/ancestor admission failure.
189        source: ExecutionWorkflowPersistenceError,
190        /// Existing durable publisher's returned checksum.
191        checksum: ArtifactChecksumRecord,
192    },
193}
194
195/// Rejection of a returned data reply while retaining its original attempt.
196#[derive(Debug, Error)]
197pub enum IcSnapshotDownloadReplyError<E: std::error::Error + 'static> {
198    /// Fresh authenticated exact attribution was not qualified by the integration.
199    #[error("snapshot download reply qualification failed: {0}")]
200    Qualification(#[source] E),
201    /// Require the explicit exact original Applied receipt, without outcome inference.
202    #[error("snapshot download requires an explicit original Applied receipt")]
203    ReceiptRequired,
204    /// Existing original stage admission.
205    #[error(transparent)]
206    Stage(#[from] ExecutionWorkflowPersistenceError),
207    /// Original selected journal/spending/receipt admission.
208    #[error(transparent)]
209    Journal(#[from] AttemptJournalError),
210    /// Exact local bytes/custody admission failed.
211    #[error(transparent)]
212    Artifact(#[from] IcSnapshotArtifactError),
213}
214
215#[cfg(test)]
216mod tests;
ic_backup/workflow/ic_snapshot_download/mod.rs

ic_backup/workflow/ic_snapshot_download/
mod.rs
