ic_backup/workflow/ic_snapshot_download/

mod.rs

//! Complete original planned data transfer with explicit integration-qualified receipts.

use crate::model::data_stage_plan::SnapshotDataPlanningError;
use crate::{
    model::{
        artifacts::ArtifactChecksumRecord,
        attempt_journal::MutationReceiptRequest,
        ic_snapshot_download::IcSnapshotDownloadPlan,
        ic_snapshot_transfer_read::{
            IcSnapshotTransferReadPayload, IcSnapshotTransferReadRequest,
            IcSnapshotTransferReadResponse,
        },
    },
    ops::persistence::{
        AttemptJournalError, AttemptJournalGuard, ExecutionProgressPersistenceError,
        ExecutionStageGuard, ExecutionWorkflowPersistenceError, IcSnapshotArtifactError,
        IcSnapshotArtifactWriter, read_execution_progress,
    },
    policy::ic_snapshot_transfer_read::{IcSnapshotTransferReadReply, IcSnapshotTransferReadView},
    ports::ic_snapshot_transfer_read::IcSnapshotTransferReadProvider,
    workflow::ic_snapshot_transfer_read::{
        IcSnapshotTransferReadExecutionError, read_snapshot_with,
    },
};
use thiserror::Error;

/// Stream every original data request into an existing private writer and publish it.
///
/// The metadata read and its independently qualified receipt/checkpoint must already
/// exist. Prepare the metadata-derived data stage and its complete original journals,
/// retain exact token/raw-ID association, then create the writer with exact raw metadata.
/// Hold no attempt guards on entry. Read-free plans and any consumed original attempt
/// reject; this is a fresh transfer, never partial-read resume or artifact recovery.
///
/// For every request, `admit` qualifies fresh access, metadata/command custody and
/// application requirements under the existing single-read coordinator. `qualify`
/// must independently authenticate exact original response attribution and return an
/// explicit original Applied receipt. Wire shape alone cannot qualify that receipt.
/// It runs under the same original journal lock held through dispatch and reuses
/// that read's decoded reply. Append exact decoded bytes before
/// the sole journal owner records that receipt; only then may the next dependent read
/// run. Failed qualification, append or persistence stops without another provider call.
/// Admission, submission and qualification are awaited under their selected journal
/// guards. Retain each reply durably before cancellable qualification work. Dropping
/// this future consumes the writer, retaining partial bytes, earlier receipts and
/// the current pending reservation; it grants no resume or repeat-call authority.
///
/// All original limits/records remain unchanged. Errors consume the writer and preserve
/// partial bytes, pending spending and every recorded receipt. Returned bounded replies
/// survive post-read rejection. Success uses the existing fresh checksum/durable
/// publisher, but supplies no immutable manifest/checkpoint, complete product backup,
/// terminal proof, default provider, refund or fence/reference release. Stable bytes,
/// complete authenticated backend transfer and application safety stay integration-owned.
/// # Errors
/// Rejects changed stage/plan/metadata/writer, prior consumption, spending, admission,
/// provider or receipt failures, malformed replies and local IO/publication failure.
pub async fn download_snapshot<E: std::error::Error + 'static>(
    stage: &ExecutionStageGuard<'_>,
    download: &IcSnapshotDownloadPlan<'_, '_>,
    mut writer: IcSnapshotArtifactWriter<'_, '_, '_>,
    provider: &mut impl IcSnapshotTransferReadProvider,
    mut admit: impl AsyncFnMut(&IcSnapshotTransferReadRequest<'_, '_>) -> Result<(), E>,
    mut qualify: impl AsyncFnMut(
        &IcSnapshotTransferReadRequest<'_, '_>,
        &IcSnapshotTransferReadResponse,
    ) -> Result<MutationReceiptRequest, E>,
) -> Result<ArtifactChecksumRecord, IcSnapshotDownloadExecutionError<E>> {
    download.validate_binding(stage.binding())?;
    let plan = download
        .plan()
        .ok_or(SnapshotDataPlanningError::NoDataOperations)?;
    if stage.plan() != plan
        || !download.matches_metadata(writer.coverage().metadata())
        || writer.coverage().covered_region_bytes() != [0; 3]
        || writer.coverage().covered_chunks() != 0
    {
        return Err(IcSnapshotDownloadExecutionError::OriginalMismatch);
    }
    let intent = plan.digest();
    writer.validate_transfer_origin(stage.layout()?.root(), intent.hash())?;
    let progress = read_execution_progress(stage.layout()?, &intent)?;
    if !progress.is_unattempted() {
        return Err(IcSnapshotDownloadExecutionError::AlreadyAttempted);
    }
    // The plan constructor pairs each request with one ordered operation.
    for (operation, payload) in plan.operations().iter().zip(download.requests()) {
        let sequence = operation.operation_sequence();
        let payload = payload.map_err(SnapshotDataPlanningError::from)?;
        let (response, result) = read_snapshot_with(
            stage,
            sequence,
            IcSnapshotTransferReadPayload::Data(&payload),
            provider,
            &mut admit,
            async |journal, request, admitted| {
                append_qualified(stage, journal, request, admitted, writer, &mut qualify).await
            },
        )
        .await?;
        writer = result.map_err(|source| IcSnapshotDownloadExecutionError::AfterReply {
            operation_sequence: sequence,
            source,
            response: Box::new(response),
        })?;
    }
    let progress = read_execution_progress(stage.layout()?, &intent)?;
    if !progress.all_applied() {
        return Err(IcSnapshotDownloadExecutionError::AlreadyAttempted);
    }
    writer.validate_transfer_origin(stage.layout()?.root(), intent.hash())?;
    let checksum = writer.finish()?;
    if let Err(source) = stage.layout() {
        return Err(IcSnapshotDownloadExecutionError::AfterPublication { source, checksum });
    }
    Ok(checksum)
}

async fn append_qualified<'journal, 'layout, 'metadata, E: std::error::Error + 'static>(
    stage: &ExecutionStageGuard<'_>,
    journal: &mut AttemptJournalGuard<'_>,
    request: &IcSnapshotTransferReadRequest<'_, '_>,
    admitted: &IcSnapshotTransferReadView<'_, '_>,
    writer: IcSnapshotArtifactWriter<'journal, 'layout, 'metadata>,
    qualify: &mut impl AsyncFnMut(
        &IcSnapshotTransferReadRequest<'_, '_>,
        &IcSnapshotTransferReadResponse,
    ) -> Result<MutationReceiptRequest, E>,
) -> Result<IcSnapshotArtifactWriter<'journal, 'layout, 'metadata>, IcSnapshotDownloadReplyError<E>>
{
    let receipt = qualify(request, admitted.response())
        .await
        .map_err(IcSnapshotDownloadReplyError::Qualification)?;
    let writer = super::qualified_receipt::record_applied(
        journal,
        receipt,
        request.mutation_attempt(),
        request.payload().digest().hash(),
        IcSnapshotDownloadReplyError::ReceiptRequired,
        || {
            // The sealed request authority already binds this admitted stage plan.
            writer.validate_transfer_origin(
                stage.layout()?.root(),
                request.authority().binding().intent(),
            )?;
            let IcSnapshotTransferReadReply::Data(reply) = admitted.reply() else {
                return Err(IcSnapshotDownloadReplyError::ReceiptRequired);
            };
            Ok(writer.append(reply)?)
        },
    )?;
    stage.layout()?;
    Ok(writer)
}

/// Failures preserve original spending and partial/published artifact evidence.
#[derive(Debug, Error)]
pub enum IcSnapshotDownloadExecutionError<E: std::error::Error + 'static> {
    /// Exact retained writer, stage or metadata differs.
    #[error("snapshot download originals differ")]
    OriginalMismatch,
    /// A prior attempt cannot be replayed by this fresh transfer entrypoint.
    #[error("snapshot download original data stage was already attempted")]
    AlreadyAttempted,
    /// Existing original download plan/binding admission.
    #[error(transparent)]
    Planning(#[from] SnapshotDataPlanningError),
    /// Existing retained stage/ancestor admission.
    #[error(transparent)]
    Stage(#[from] ExecutionWorkflowPersistenceError),
    /// Complete original progress admission.
    #[error(transparent)]
    Progress(#[from] ExecutionProgressPersistenceError),
    /// One original read failed; its owner retains any returned bounded reply.
    #[error(transparent)]
    Read(#[from] IcSnapshotTransferReadExecutionError<E>),
    /// Canonical writer/checksum/durable publication failure.
    #[error(transparent)]
    Artifact(#[from] IcSnapshotArtifactError),
    /// Rejection after a returned read retains the exact bounded response.
    #[error("snapshot download reply rejected for operation {operation_sequence}: {source}")]
    AfterReply {
        /// Exact original data operation.
        operation_sequence: u64,
        /// Original rejection without undoing receipt or spending.
        source: IcSnapshotDownloadReplyError<E>,
        /// Exact bounded returned bytes and passive claims.
        response: Box<IcSnapshotTransferReadResponse>,
    },
    /// Publication succeeded but stage re-admission failed; retain the published checksum.
    #[error("snapshot download stage changed after publication: {source}")]
    AfterPublication {
        /// Retained stage/ancestor admission failure.
        source: ExecutionWorkflowPersistenceError,
        /// Existing durable publisher's returned checksum.
        checksum: ArtifactChecksumRecord,
    },
}

/// Rejection of a returned data reply while retaining its original attempt.
#[derive(Debug, Error)]
pub enum IcSnapshotDownloadReplyError<E: std::error::Error + 'static> {
    /// Fresh authenticated exact attribution was not qualified by the integration.
    #[error("snapshot download reply qualification failed: {0}")]
    Qualification(#[source] E),
    /// Require the explicit exact original Applied receipt, without outcome inference.
    #[error("snapshot download requires an explicit original Applied receipt")]
    ReceiptRequired,
    /// Existing original stage admission.
    #[error(transparent)]
    Stage(#[from] ExecutionWorkflowPersistenceError),
    /// Original selected journal/spending/receipt admission.
    #[error(transparent)]
    Journal(#[from] AttemptJournalError),
    /// Exact local bytes/custody admission failed.
    #[error(transparent)]
    Artifact(#[from] IcSnapshotArtifactError),
}

#[cfg(test)]
mod tests;

impl<E: std::error::Error + 'static> IcSnapshotDownloadExecutionError<E> {
    /// Borrow the bounded returned evidence carried by this error, including nested failures.
    /// Absence describes retained evidence only and grants no dispatch or retry authority.
    #[must_use]
    pub fn response(&self) -> Option<&IcSnapshotTransferReadResponse> {
        match self {
            Self::AfterReply { response, .. } => Some(response),
            Self::Read(source) => source.response(),
            _ => None,
        }
    }
}