ic_backup/workflow/ic_snapshot_metadata/

mod.rs

//! One original metadata stage, with explicit qualified receipt and learned checkpoint.

use crate::{
    model::{
        artifacts::ArtifactChecksumRecord,
        attempt_journal::MutationReceiptRequest,
        execution_workflow::ExecutionStagePredecessorRecord,
        ic_snapshot_metadata::IcSnapshotMetadataRequest,
        ic_snapshot_transfer_read::{
            IcSnapshotTransferReadPayload, IcSnapshotTransferReadRequest,
            IcSnapshotTransferReadResponse,
        },
    },
    ops::persistence::{
        AttemptJournalError, AttemptJournalGuard, ExecutionStageCheckpointError,
        ExecutionStageGuard, ExecutionWorkflowPersistenceError,
    },
    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;

/// Read one new exact singleton metadata stage, record its qualified receipt and checkpoint.
///
/// Prepare the original stage and complete journals first; hold no attempt guards on
/// entry. Reuse the canonical single-read owner for durable spending, mandatory fresh
/// admission and exactly one provider call. A pending or previously Applied read is
/// never dispatched again, including after interruption before checkpoint publication.
///
/// `admit` owns fresh read permission, raw-ID/application and never-dispatched custody.
/// `qualify` runs under the original journal lock and must independently authenticate
/// exact response attribution AND durably retain original request/reply bytes before
/// returning the explicit Applied receipt. There is no default qualification or byte
/// store. Opaque provider evidence and valid wire shape cannot substitute for it.
/// Account any remote qualification separately before its calls.
/// Admission, submission and independent qualification are async. Qualification
/// holds the same original journal continuously from dispatch, reuses its decoded
/// reply, and must retain the reply durably before
/// awaiting cancellable work. Cancellation keeps pending/Applied spending and denies
/// reentry; no executor or `Send` bound is imposed by the core.
///
/// Record the receipt through the sole spending owner, release its lock, then publish
/// the canonical all-Applied checkpoint with the exact request/raw-metadata digest as
/// learned evidence. The returned response and predecessor can feed the existing
/// metadata decoder and `IcSnapshotDownloadPlan::bind`; preparing the data stage and
/// writer remains explicit. Failure retains spending, receipts, checkpoints and bytes;
/// post-read errors retain the bounded response. Reopen/checkpoint recovery never
/// repeats a read. No data calls, artifact, default Agent bridge, complete backup,
/// terminal proof or fence/reference release follows.
/// # Errors
/// Rejects non-singleton or changed originals, spending, fresh admission, provider,
/// association, qualification, receipt and checkpoint failures without cleanup/refund.
pub async fn read_snapshot_metadata<E: std::error::Error + 'static>(
    stage: &ExecutionStageGuard<'_>,
    operation_sequence: u64,
    payload: &IcSnapshotMetadataRequest,
    provider: &mut impl IcSnapshotTransferReadProvider,
    admit: impl AsyncFnOnce(&IcSnapshotTransferReadRequest<'_, '_>) -> Result<(), E>,
    qualify: impl AsyncFnOnce(
        &IcSnapshotTransferReadRequest<'_, '_>,
        &IcSnapshotTransferReadResponse,
    ) -> Result<MutationReceiptRequest, E>,
) -> Result<
    (
        IcSnapshotTransferReadResponse,
        ExecutionStagePredecessorRecord,
    ),
    IcSnapshotMetadataExecutionError<E>,
> {
    let operations = stage.plan().operations();
    if operations.len() != 1
        || operations[0].operation_sequence() != operation_sequence
        || operations[0].request() != payload.digest().hash()
    {
        return Err(IcSnapshotMetadataExecutionError::OriginalMismatch);
    }
    let (response, settlement) = read_snapshot_with(
        stage,
        operation_sequence,
        IcSnapshotTransferReadPayload::Metadata(payload),
        provider,
        admit,
        async |journal, request, admitted| {
            settle_metadata(stage, journal, request, admitted, qualify).await
        },
    )
    .await?;
    match settlement.and_then(|digest| Ok(stage.reconcile_checkpoint(digest)?)) {
        Ok(predecessor) => Ok((response, predecessor)),
        Err(source) => Err(IcSnapshotMetadataExecutionError::AfterReply {
            source,
            response: Box::new(response),
        }),
    }
}

async fn settle_metadata<E: std::error::Error + 'static>(
    stage: &ExecutionStageGuard<'_>,
    journal: &mut AttemptJournalGuard<'_>,
    request: &IcSnapshotTransferReadRequest<'_, '_>,
    admitted: &IcSnapshotTransferReadView<'_, '_>,
    qualify: impl AsyncFnOnce(
        &IcSnapshotTransferReadRequest<'_, '_>,
        &IcSnapshotTransferReadResponse,
    ) -> Result<MutationReceiptRequest, E>,
) -> Result<ArtifactChecksumRecord, IcSnapshotMetadataSettlementError<E>> {
    let receipt = qualify(request, admitted.response())
        .await
        .map_err(IcSnapshotMetadataSettlementError::Qualification)?;
    let receipt = super::qualified_receipt::admit_applied(
        receipt,
        request.mutation_attempt(),
        request.payload().digest().hash(),
    )
    .ok_or(IcSnapshotMetadataSettlementError::ReceiptRequired)?;
    let digest = match admitted.reply() {
        IcSnapshotTransferReadReply::Metadata(metadata) => metadata.digest(),
        IcSnapshotTransferReadReply::Data(_) => {
            return Err(IcSnapshotMetadataSettlementError::ReceiptRequired);
        }
    };
    stage.layout()?;
    journal.record_mutation(receipt)?;
    Ok(digest)
}

/// Metadata coordination failures retain exact original spending and returned evidence.
#[derive(Debug, Error)]
pub enum IcSnapshotMetadataExecutionError<E: std::error::Error + 'static> {
    /// The original plan is not exactly this single metadata operation.
    #[error("snapshot metadata requires an exact singleton original stage")]
    OriginalMismatch,
    /// Canonical single-read failure, including any bounded returned response.
    #[error(transparent)]
    Read(#[from] IcSnapshotTransferReadExecutionError<E>),
    /// Receipt or checkpoint failed after a response; all originals remain retained.
    #[error("snapshot metadata reply settlement failed: {source}")]
    AfterReply {
        /// Exact original rejection; no retry or refund follows.
        source: IcSnapshotMetadataSettlementError<E>,
        /// Original bounded reply and passive claims.
        response: Box<IcSnapshotTransferReadResponse>,
    },
}

/// Rejection after one returned metadata reply; no second accounting owner.
#[derive(Debug, Error)]
pub enum IcSnapshotMetadataSettlementError<E: std::error::Error + 'static> {
    /// Actual authentication, attribution or durable original-byte retention failed.
    #[error("snapshot metadata qualification failed: {0}")]
    Qualification(#[source] E),
    /// Require an independently qualified exact original Applied receipt.
    #[error("snapshot metadata requires an explicit original Applied receipt")]
    ReceiptRequired,
    /// Original stage or retained ancestor admission failed.
    #[error(transparent)]
    Stage(#[from] ExecutionWorkflowPersistenceError),
    /// Original selected journal or receipt admission failed.
    #[error(transparent)]
    Journal(#[from] AttemptJournalError),
    /// Canonical all-Applied checkpoint publication/re-admission failed.
    #[error(transparent)]
    Checkpoint(#[from] ExecutionStageCheckpointError),
}

#[cfg(all(test, unix))]
mod tests;