ic-backup 0.44.1

//! Complete finite original data uploads with explicit integration-qualified receipts.

use super::{IcSnapshotUploadExecutionError, upload_snapshot_with};
use crate::{
    model::{
        attempt_journal::MutationReceiptRequest,
        evidence_digest::EvidenceDigest,
        execution_workflow::ExecutionStagePredecessorRecord,
        ic_mutation::IcMutationAcknowledgement,
        ic_snapshot_upload::{
            IcSnapshotDataUploadPlan, IcSnapshotDataUploadPlanningError, IcSnapshotUploadAttempt,
        },
    },
    ops::persistence::{
        AttemptJournalError, DownloadJournalGuard, ExecutionProgressPersistenceError,
        ExecutionStageCheckpointError, ExecutionStageGuard, ExecutionWorkflowPersistenceError,
        IcSnapshotUploadArtifactError, read_execution_progress,
    },
    ports::ic_snapshot_upload::IcSnapshotUploadProvider,
};
use thiserror::Error;

/// Upload every exact original data operation, then checkpoint qualified reply evidence.
///
/// Prepare the source-derived data plan, allocation predecessor, retained stage and
/// complete original journals first; hold no attempt guards. Empty plans and any prior
/// consumption reject before dispatch: this is a fresh stage, never partial upload
/// resume. Original source and destination/command custody remain integration-owned.
/// Fresh guarded preparation buffers one payload and rechecks its immutable binding
/// before existing single-upload reservation, mandatory fresh `admit` and one call.
///
/// `qualify` must independently authenticate exact original data-write attribution and
/// durably retain original request/reply bytes before returning its explicit Applied
/// receipt under the selected journal lock. An empty reply never supplies a receipt.
/// Account remote qualification separately before its calls. Record through the sole
/// attempt owner before the next dependency can dispatch; failures stop immediately.
/// Admission, submission and qualification are async under their selected journal
/// guards. Retain each exact reply before awaiting cancellable qualification work.
/// Cancellation preserves source bytes/references, earlier Applied receipts and
/// the current pending reservation, denying whole-stage or single-call reentry.
///
/// After all exact receipts, release journal locks and checkpoint the full original plan
/// and ordered canonical request/raw-reply evidence through the existing stage owner.
/// This is declared stage settlement, not authenticated whole-backend completeness,
/// same-release load/start admission, terminal proof or fence/source-reference release.
/// No hidden observation, reissue, refund, default provider or journal is added.
/// # Errors
/// Rejects changed originals, consumed attempts, source/preparation failures, spending,
/// admission/provider/reply/qualification/receipt or checkpoint failure. Bounded returned
/// replies survive post-reply rejection; spending, recorded receipts and source remain.
pub async fn upload_snapshot_data<E: std::error::Error + 'static>(
    stage: &ExecutionStageGuard<'_>,
    upload: &IcSnapshotDataUploadPlan<'_, '_, '_>,
    source: &DownloadJournalGuard<'_>,
    snapshot: &str,
    provider: &mut impl IcSnapshotUploadProvider,
    mut admit: impl AsyncFnMut(&IcSnapshotUploadAttempt<'_, '_>) -> Result<(), E>,
    mut qualify: impl AsyncFnMut(
        &IcSnapshotUploadAttempt<'_, '_>,
        &IcMutationAcknowledgement,
    ) -> Result<MutationReceiptRequest, E>,
) -> Result<ExecutionStagePredecessorRecord, IcSnapshotDataUploadExecutionError<E>> {
    upload.validate_binding(stage.binding())?;
    let plan = upload
        .plan()
        .ok_or(IcSnapshotDataUploadPlanningError::NoDataWrites)?;
    if stage.plan() != plan {
        return Err(IcSnapshotDataUploadPlanningError::OriginalMismatch.into());
    }
    let progress = read_execution_progress(stage.layout()?, &plan.digest())?;
    if progress.attempts.mutations_used != 0 || progress.attempts.observations_used != 0 {
        return Err(IcSnapshotDataUploadExecutionError::AlreadyAttempted);
    }
    let mut evidence = EvidenceDigest::new(b"ic-backup/ic-snapshot-data-upload/v1\0");
    evidence.append(plan.digest().hash().as_bytes());
    for (index, kind) in upload.kinds().iter().enumerate() {
        let sequence =
            u64::try_from(index).map_err(|_| IcSnapshotDataUploadPlanningError::CountOverflow)?;
        let metadata = upload.metadata();
        let payload = source.prepare_ic_snapshot_upload_data(
            metadata.source_plan(),
            snapshot,
            metadata,
            upload.destination(),
            kind.clone(),
        )?;
        // The single-call owner rechecks the exact full source/context/request before spending.
        let (acknowledgement, settlement) = upload_snapshot_with(
            stage,
            sequence,
            &payload,
            provider,
            &mut admit,
            async |journal, request, admitted| {
                super::super::qualified_receipt::AppliedReceiptRecorder {
                    journal,
                    expected_attempt: request.mutation_attempt(),
                    expected_request: payload.binding_digest().hash(),
                }
                .record_applied(
                    request,
                    admitted.acknowledgement(),
                    async |request, reply| qualify(request, reply).await,
                    IcSnapshotDataUploadReplyError::Qualification,
                    || IcSnapshotDataUploadReplyError::ReceiptRequired,
                    || {
                        stage
                            .layout()
                            .map(|_| ())
                            .map_err(IcSnapshotDataUploadReplyError::Stage)
                    },
                )
                .await?;
                stage.layout()?;
                Ok::<_, IcSnapshotDataUploadReplyError<E>>(admitted.reply().digest())
            },
        )
        .await?;
        let digest =
            settlement.map_err(|source| IcSnapshotDataUploadExecutionError::AfterReply {
                operation_sequence: sequence,
                source,
                acknowledgement: Box::new(acknowledgement),
            })?;
        evidence.append(digest.hash().as_bytes());
    }
    let progress = read_execution_progress(stage.layout()?, &plan.digest())?;
    if progress.applied_operations != upload.kinds().len() {
        return Err(IcSnapshotDataUploadExecutionError::AlreadyAttempted);
    }
    Ok(stage.reconcile_checkpoint(evidence.finish())?)
}

/// Data-stage failure preserves source, original spending and returned acknowledgement bytes.
#[derive(Debug, Error)]
pub enum IcSnapshotDataUploadExecutionError<E: std::error::Error + 'static> {
    /// Fresh-only entrypoint rejects any original consumed mutation or observation.
    #[error("snapshot data upload original stage was already attempted")]
    AlreadyAttempted,
    /// Existing exact original data-plan/binding admission.
    #[error(transparent)]
    Planning(#[from] IcSnapshotDataUploadPlanningError),
    /// Existing retained workflow/stage/ancestor admission.
    #[error(transparent)]
    Stage(#[from] ExecutionWorkflowPersistenceError),
    /// Complete original progress admission.
    #[error(transparent)]
    Progress(#[from] ExecutionProgressPersistenceError),
    /// Fresh exact source preparation failed before the selected operation reserves.
    #[error(transparent)]
    Source(#[from] IcSnapshotUploadArtifactError),
    /// One exact original upload failed; its owner retains bounded returned evidence.
    #[error(transparent)]
    Upload(#[from] IcSnapshotUploadExecutionError<E>),
    /// Receipt settlement failed after a bounded returned acknowledgement.
    #[error("snapshot data upload reply rejected for operation {operation_sequence}: {source}")]
    AfterReply {
        /// Exact original failed operation sequence.
        operation_sequence: u64,
        /// Original qualification or persistence rejection.
        source: IcSnapshotDataUploadReplyError<E>,
        /// Exact bounded reply, without an inferred outcome.
        acknowledgement: Box<IcMutationAcknowledgement>,
    },
    /// Existing all-Applied stage checkpoint failed; receipts and occupied evidence remain.
    #[error(transparent)]
    Checkpoint(#[from] ExecutionStageCheckpointError),
}

/// Post-reply data-write rejection, without a second outcome/accounting owner.
#[derive(Debug, Error)]
pub enum IcSnapshotDataUploadReplyError<E: std::error::Error + 'static> {
    /// Independent attribution/authentication or durable original-byte retention failed.
    #[error("snapshot data upload qualification failed: {0}")]
    Qualification(#[source] E),
    /// Require the explicit exact original Applied receipt.
    #[error("snapshot data upload requires an explicit original Applied receipt")]
    ReceiptRequired,
    /// Original workflow/stage/ancestor admission failed.
    #[error(transparent)]
    Stage(#[from] ExecutionWorkflowPersistenceError),
    /// Existing selected journal/receipt persistence failed.
    #[error(transparent)]
    Journal(#[from] AttemptJournalError),
}