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.matches_metadata(writer.coverage().metadata())
73        || writer.coverage().covered_region_bytes() != [0; 3]
74        || writer.coverage().covered_chunks() != 0
75    {
76        return Err(IcSnapshotDownloadExecutionError::OriginalMismatch);
77    }
78    writer.validate_transfer_origin(stage.layout()?.root(), plan.digest().hash())?;
79    let progress = read_execution_progress(stage.layout()?, &plan.digest())?;
80    if progress.attempts.mutations_used != 0 || progress.attempts.observations_used != 0 {
81        return Err(IcSnapshotDownloadExecutionError::AlreadyAttempted);
82    }
83    for (index, payload) in download.requests().iter().enumerate() {
84        // Planning already bounded the ordinal by the original stage allowance.
85        let sequence =
86            u64::try_from(index).map_err(|_| IcSnapshotDownloadPlanningError::CountOverflow)?;
87        let (response, result) = read_snapshot_with(
88            stage,
89            sequence,
90            IcSnapshotTransferReadPayload::Data(payload),
91            provider,
92            &mut admit,
93            async |journal, request, admitted| {
94                append_qualified(stage, journal, request, admitted, writer, &mut qualify).await
95            },
96        )
97        .await?;
98        writer = result.map_err(|source| IcSnapshotDownloadExecutionError::AfterReply {
99            operation_sequence: sequence,
100            source,
101            response: Box::new(response),
102        })?;
103    }
104    let progress = read_execution_progress(stage.layout()?, &plan.digest())?;
105    if progress.applied_operations != download.requests().len() {
106        return Err(IcSnapshotDownloadExecutionError::AlreadyAttempted);
107    }
108    writer.validate_transfer_origin(stage.layout()?.root(), plan.digest().hash())?;
109    let checksum = writer.finish()?;
110    if let Err(source) = stage.layout() {
111        return Err(IcSnapshotDownloadExecutionError::AfterPublication { source, checksum });
112    }
113    Ok(checksum)
114}
115
116async fn append_qualified<'journal, 'layout, 'metadata, E: std::error::Error + 'static>(
117    stage: &ExecutionStageGuard<'_>,
118    journal: &mut AttemptJournalGuard<'_>,
119    request: &IcSnapshotTransferReadRequest<'_, '_>,
120    admitted: &IcSnapshotTransferReadView<'_, '_>,
121    writer: IcSnapshotArtifactWriter<'journal, 'layout, 'metadata>,
122    qualify: &mut impl AsyncFnMut(
123        &IcSnapshotTransferReadRequest<'_, '_>,
124        &IcSnapshotTransferReadResponse,
125    ) -> Result<MutationReceiptRequest, E>,
126) -> Result<IcSnapshotArtifactWriter<'journal, 'layout, 'metadata>, IcSnapshotDownloadReplyError<E>>
127{
128    let plan = stage.plan();
129    let receipt = qualify(request, admitted.response())
130        .await
131        .map_err(IcSnapshotDownloadReplyError::Qualification)?;
132    let receipt = super::qualified_receipt::admit_applied(
133        receipt,
134        request.mutation_attempt(),
135        request.payload().digest().hash(),
136    )
137    .ok_or(IcSnapshotDownloadReplyError::ReceiptRequired)?;
138    writer.validate_transfer_origin(stage.layout()?.root(), plan.digest().hash())?;
139    let IcSnapshotTransferReadReply::Data(reply) = admitted.reply() else {
140        return Err(IcSnapshotDownloadReplyError::ReceiptRequired);
141    };
142    let writer = writer.append(reply)?;
143    journal.record_mutation(receipt)?;
144    stage.layout()?;
145    Ok(writer)
146}
147
148/// Failures preserve original spending and partial/published artifact evidence.
149#[derive(Debug, Error)]
150pub enum IcSnapshotDownloadExecutionError<E: std::error::Error + 'static> {
151    /// Exact retained writer, stage or metadata differs.
152    #[error("snapshot download originals differ")]
153    OriginalMismatch,
154    /// A prior attempt cannot be replayed by this fresh transfer entrypoint.
155    #[error("snapshot download original data stage was already attempted")]
156    AlreadyAttempted,
157    /// Existing original download plan/binding admission.
158    #[error(transparent)]
159    Planning(#[from] IcSnapshotDownloadPlanningError),
160    /// Existing retained stage/ancestor admission.
161    #[error(transparent)]
162    Stage(#[from] ExecutionWorkflowPersistenceError),
163    /// Complete original progress admission.
164    #[error(transparent)]
165    Progress(#[from] ExecutionProgressPersistenceError),
166    /// One original read failed; its owner retains any returned bounded reply.
167    #[error(transparent)]
168    Read(#[from] IcSnapshotTransferReadExecutionError<E>),
169    /// Canonical writer/checksum/durable publication failure.
170    #[error(transparent)]
171    Artifact(#[from] IcSnapshotArtifactError),
172    /// Rejection after a returned read retains the exact bounded response.
173    #[error("snapshot download reply rejected for operation {operation_sequence}: {source}")]
174    AfterReply {
175        /// Exact original data operation.
176        operation_sequence: u64,
177        /// Original rejection without undoing receipt or spending.
178        source: IcSnapshotDownloadReplyError<E>,
179        /// Exact bounded returned bytes and passive claims.
180        response: Box<IcSnapshotTransferReadResponse>,
181    },
182    /// Publication succeeded but stage re-admission failed; retain the published checksum.
183    #[error("snapshot download stage changed after publication: {source}")]
184    AfterPublication {
185        /// Retained stage/ancestor admission failure.
186        source: ExecutionWorkflowPersistenceError,
187        /// Existing durable publisher's returned checksum.
188        checksum: ArtifactChecksumRecord,
189    },
190}
191
192/// Rejection of a returned data reply while retaining its original attempt.
193#[derive(Debug, Error)]
194pub enum IcSnapshotDownloadReplyError<E: std::error::Error + 'static> {
195    /// Fresh authenticated exact attribution was not qualified by the integration.
196    #[error("snapshot download reply qualification failed: {0}")]
197    Qualification(#[source] E),
198    /// Require the explicit exact original Applied receipt, without outcome inference.
199    #[error("snapshot download requires an explicit original Applied receipt")]
200    ReceiptRequired,
201    /// Existing original stage admission.
202    #[error(transparent)]
203    Stage(#[from] ExecutionWorkflowPersistenceError),
204    /// Original selected journal/spending/receipt admission.
205    #[error(transparent)]
206    Journal(#[from] AttemptJournalError),
207    /// Exact local bytes/custody admission failed.
208    #[error(transparent)]
209    Artifact(#[from] IcSnapshotArtifactError),
210}
211
212#[cfg(test)]
213mod tests;
ic_backup/workflow/ic_snapshot_download/mod.rs

ic_backup/workflow/ic_snapshot_download/
mod.rs
