ic-backup 0.61.1

//! Exact bounded data-upload declarations, retaining no aggregate payload bytes.

use super::{IcSnapshotUploadReply, IcSnapshotUploadReplyKind, IcSnapshotUploadRequest};
use crate::model::{
    artifacts::ArtifactChecksumRecord,
    data_stage_plan::{self, LearnedDataStage, SnapshotDataPlanningError},
    execution_workflow::{
        ExecutionStageBindingRecord, ExecutionStagePredecessorRecord, ExecutionWorkflowRecord,
    },
    ic_snapshot_data::planned_extents,
    operation_plan::{OperationPlanRecord, PlannedOperationRecord},
};
use ic_platform::management_canister_types::SnapshotDataKind;

/// Complete ordered extents and source-bound request digests for one original data stage.
///
/// Guarded local source preparation creates this declaration. Module, heap and stable
/// regions precede known chunks in original metadata order. Every operation has one
/// update and zero observations with an explicit predecessor dependency. Original
/// stage headroom stays unassigned. No upload bytes, spending, authenticated allocation
/// or complete-transfer proof are retained here. Empty data needs no placeholder plan.
pub struct IcSnapshotDataUploadPlan<'workflow, 'request, 'source> {
    binding: LearnedDataStage<'workflow>,
    metadata: &'request IcSnapshotUploadRequest<'source>,
    destination: Vec<u8>,
    writes: Option<DataWrites>,
}

// Empty data has neither extents nor a placeholder plan; writes retain both together.
struct DataWrites {
    kinds: Vec<SnapshotDataKind>,
    plan: OperationPlanRecord,
}
// Completion records the admitted operation, never a new effect or persisted mode.
#[derive(Clone)]
pub(crate) enum UploadCompletion {
    Data,
    Allocation {
        sequence: u64,
        evidence: ArtifactChecksumRecord,
    },
}

// Carry the admitted allocation and bounded extents across guarded source IO.
// Only finish can turn their exact ordered byte bindings into a public plan.
pub(crate) struct DataUploadPreparation<'workflow, 'request, 'source> {
    workflow: &'workflow ExecutionWorkflowRecord,
    stage: &'workflow PlannedOperationRecord,
    metadata: &'request IcSnapshotUploadRequest<'source>,
    destination: Vec<u8>,
    allocation_evidence: ArtifactChecksumRecord,
    kinds: Vec<SnapshotDataKind>,
}

impl<'workflow, 'request, 'source> DataUploadPreparation<'workflow, 'request, 'source> {
    pub(crate) fn new(
        workflow: &'workflow ExecutionWorkflowRecord,
        sequence: u64,
        allocation: &IcSnapshotUploadReply<'request, 'source>,
        chunk_bytes: u64,
    ) -> Result<Self, SnapshotDataPlanningError> {
        let metadata = allocation.request();
        let stage = workflow.stage(sequence)?;
        let IcSnapshotUploadReplyKind::Metadata { snapshot_id } = allocation.kind() else {
            return Err(SnapshotDataPlanningError::OriginalMismatch);
        };
        metadata.validate_data_destination(snapshot_id)?;
        if stage.target() != metadata.target()
            || !super::attempt::source_context_matches(workflow.allocation(), metadata)
        {
            return Err(SnapshotDataPlanningError::OriginalMismatch);
        }
        let kinds = planned_extents(metadata.source(), chunk_bytes, stage.budget().mutations())?;
        Ok(Self {
            workflow,
            stage,
            metadata,
            destination: snapshot_id.clone(),
            allocation_evidence: allocation.digest(),
            kinds,
        })
    }

    pub(crate) const fn metadata(&self) -> &'request IcSnapshotUploadRequest<'source> {
        self.metadata
    }
    pub(crate) fn destination(&self) -> &[u8] {
        &self.destination
    }
    pub(crate) fn kinds(&self) -> &[SnapshotDataKind] {
        &self.kinds
    }

    pub(crate) fn finish(
        self,
        bindings: Vec<ArtifactChecksumRecord>,
    ) -> Result<IcSnapshotDataUploadPlan<'workflow, 'request, 'source>, SnapshotDataPlanningError>
    {
        if bindings.len() != self.kinds.len() {
            return Err(SnapshotDataPlanningError::OriginalMismatch);
        }
        let plan =
            data_stage_plan::build(self.workflow.allocation(), self.stage, bindings.into_iter())?;
        Ok(IcSnapshotDataUploadPlan {
            binding: LearnedDataStage {
                workflow: self.workflow,
                sequence: self.stage.operation_sequence(),
                evidence: self.allocation_evidence,
            },
            metadata: self.metadata,
            destination: self.destination,
            writes: plan.map(|plan| DataWrites {
                kinds: self.kinds,
                plan,
            }),
        })
    }
}

impl<'request, 'source> IcSnapshotDataUploadPlan<'_, 'request, 'source> {
    pub(crate) fn workflow_digest(&self) -> ArtifactChecksumRecord {
        self.binding.workflow.digest()
    }
    pub(crate) fn validate_completion(
        &self,
        binding: &ExecutionStageBindingRecord,
        plan: &OperationPlanRecord,
        learned: &ArtifactChecksumRecord,
    ) -> Result<UploadCompletion, SnapshotDataPlanningError> {
        if let Some(expected) = self.plan() {
            self.validate_binding(binding)?;
            if plan != expected {
                return Err(SnapshotDataPlanningError::OriginalMismatch);
            }
            Ok(UploadCompletion::Data)
        } else {
            binding.validate(self.binding.workflow, plan)?;
            let [operation] = plan.operations() else {
                return Err(SnapshotDataPlanningError::OriginalMismatch);
            };
            if operation.target() != self.metadata.target()
                || operation.request() != self.metadata.binding_digest().hash()
                || learned != &self.binding.evidence
            {
                return Err(SnapshotDataPlanningError::OriginalMismatch);
            }
            Ok(UploadCompletion::Allocation {
                sequence: operation.operation_sequence(),
                evidence: self.binding.evidence.clone(),
            })
        }
    }
    /// Read exact ordered source extents. Operation sequences are zero-based indices.
    #[must_use]
    pub fn kinds(&self) -> &[SnapshotDataKind] {
        self.writes
            .as_ref()
            .map_or(&[], |writes| writes.kinds.as_slice())
    }
    /// Read the original metadata allocation declaration and retained source association.
    #[must_use]
    pub const fn metadata(&self) -> &'request IcSnapshotUploadRequest<'source> {
        self.metadata
    }
    /// Read exact declared destination bytes; they grant no allocation authority.
    #[must_use]
    pub fn destination(&self) -> &[u8] {
        &self.destination
    }
    /// Read the immutable child plan, or None when there are no data writes.
    #[must_use]
    pub const fn plan(&self) -> Option<&OperationPlanRecord> {
        match &self.writes {
            None => None,
            Some(writes) => Some(&writes.plan),
        }
    }

    /// Bind the exact original singleton allocation and request/raw-reply evidence.
    ///
    /// Existing stage persistence requires complete Applied predecessor journals and
    /// their chronological checkpoint. This check authenticates neither allocation
    /// attribution nor byte custody and grants no dispatch/load/start authority.
    /// # Errors
    /// Rejects an empty plan, changed allocation/request/evidence or original dependency.
    pub fn bind(
        &self,
        allocation_binding: &ExecutionStageBindingRecord,
        allocation_plan: &OperationPlanRecord,
        predecessors: Vec<ExecutionStagePredecessorRecord>,
    ) -> Result<ExecutionStageBindingRecord, SnapshotDataPlanningError> {
        let plan = self
            .plan()
            .ok_or(SnapshotDataPlanningError::NoDataOperations)?;
        self.binding.bind(
            plan,
            allocation_binding,
            allocation_plan,
            self.metadata.binding_digest().hash(),
            predecessors,
        )
    }
    pub(crate) fn validate_binding(
        &self,
        binding: &ExecutionStageBindingRecord,
    ) -> Result<(), SnapshotDataPlanningError> {
        let plan = self
            .plan()
            .ok_or(SnapshotDataPlanningError::NoDataOperations)?;
        self.binding.validate(binding, plan)
    }
}

#[cfg(test)]
mod tests;