ic-backup 0.64.0

//! Single opaque update or original recovery observation under existing stage authority.
//!
//! Applications retain protocol interpretation, fresh authority and durable byte custody.
//! Host ingress receipts never settle application-owned paid effects or terminal release.

use crate::{
    model::{
        application_ingress::{
            ApplicationIngressError, ApplicationIngressPayload, ApplicationMutationRequest,
            ApplicationObservationRequest,
        },
        artifacts::ArtifactChecksumRecord,
        attempt_journal::{
            AttemptJournalRecord, MutationOutcomeRecord, MutationReceiptRequest,
            ObservationOutcomeRecord, ObservationReceiptRequest,
        },
        operation_plan::OperationPlanError,
    },
    ops::persistence::{AttemptJournalError, ExecutionStageCheckpointError, ExecutionStageGuard},
    workflow::{ExecutionReservationError, execution_workflow::RetainedStageEvidence},
};
use std::future::Future;

/// Exact already-reserved request; constructing a request never authorizes redispatch.
#[derive(Debug)]
pub enum ApplicationRequest<'a> {
    /// One new original mutation.
    Mutation(ApplicationMutationRequest<'a>),
    /// One separately reserved observation of the original pending mutation.
    Observation(ApplicationObservationRequest<'a>),
}
impl ApplicationRequest<'_> {
    /// Read the exact bounded application bytes.
    #[must_use]
    pub const fn payload(&self) -> &ApplicationIngressPayload {
        match self {
            Self::Mutation(r) => r.payload(),
            Self::Observation(r) => r.payload(),
        }
    }
    fn validate(&self, journal: &AttemptJournalRecord) -> Result<(), ApplicationIngressError> {
        match self {
            Self::Mutation(r) => r.validate_journal(journal),
            Self::Observation(r) => r.validate_journal(journal),
        }
    }
    fn record_qualified<E: std::error::Error + 'static>(
        &self,
        journal: &mut crate::ops::persistence::AttemptJournalGuard<'_>,
        qualified: &ApplicationQualification,
    ) -> Result<(), ApplicationExecutionCause<E>> {
        match self {
            ApplicationRequest::Mutation(r) => {
                if qualified.outcome != ObservationOutcomeRecord::Applied {
                    return Err(ApplicationExecutionCause::MutationOutcome);
                }
                super::qualified_receipt::record_applied(
                    journal,
                    MutationReceiptRequest {
                        attempt: r.mutation_attempt(),
                        request: self.payload().digest(),
                        outcome: MutationOutcomeRecord::Applied,
                        evidence: qualified.evidence.clone(),
                    },
                    r.mutation_attempt(),
                    self.payload().digest().hash(),
                    ApplicationExecutionCause::MutationOutcome,
                    || Ok(()),
                )?;
            }
            ApplicationRequest::Observation(r) => {
                journal.record_observation(ObservationReceiptRequest {
                    attempt: r.observation_attempt(),
                    request: self.payload().digest(),
                    outcome: qualified.outcome,
                    evidence: qualified.evidence.clone(),
                })?;
            }
        }
        Ok(())
    }
}

/// Owned bounded reply, including any transport-specific evidence retained by its type.
pub trait ApplicationReply: std::fmt::Debug + 'static {
    /// Read raw bytes without application decoding.
    fn bytes(&self) -> &[u8];
}
impl ApplicationReply for Vec<u8> {
    fn bytes(&self) -> &[u8] {
        self
    }
}

/// Independently qualified application claims; matching bytes alone supply none.
#[derive(Debug)]
pub struct ApplicationQualification {
    /// Mutation replies must be Applied; observations may explicitly remain uncertain.
    pub outcome: ObservationOutcomeRecord,
    /// Durable authenticated original-request/reply attribution evidence.
    pub evidence: ArtifactChecksumRecord,
    /// Application-qualified learned input for the existing stage checkpoint.
    pub learned: ArtifactChecksumRecord,
}

impl ApplicationQualification {
    fn checkpoint(
        self,
        stage: &ExecutionStageGuard<'_>,
    ) -> Result<
        (ObservationOutcomeRecord, Option<RetainedStageEvidence>),
        ExecutionStageCheckpointError,
    > {
        let retained = if self.outcome == ObservationOutcomeRecord::Applied {
            stage.reconcile_checkpoint(self.learned.clone())?;
            Some(RetainedStageEvidence::new(
                stage.binding().digest(),
                self.learned,
            ))
        } else {
            None
        };
        Ok((self.outcome, retained))
    }
}

/// Mandatory application authority, single-dispatch transport and reply qualification.
///
/// Remote admission/qualification work requires its own separately accounted authority.
/// `submit` must durably retain original signed ingress before dispatch and exact replies
/// before returning, including before any cancellable qualification. No method may retry
/// or infer a receipt. Use Agent's existing application-ingress provider for transport.
pub trait ApplicationIngressProvider {
    /// Typed application/transport refusal; an uncertain failure grants no resend.
    type Error: std::error::Error + 'static;
    /// Exact owned reply and transport evidence.
    type Reply: ApplicationReply;
    /// Re-admit original application custody synchronously, including after each await.
    /// # Errors
    /// Refuses changed authority, byte custody or original application intent.
    fn check(&mut self, request: &ApplicationRequest<'_>) -> Result<(), Self::Error>;
    /// Qualify current membership, permissions, safety and never-dispatched custody.
    fn admit(
        &mut self,
        request: &ApplicationRequest<'_>,
    ) -> impl Future<Output = Result<(), Self::Error>>;
    /// Dispatch exactly once through the selected already-reserved lane.
    fn submit(
        &mut self,
        request: &ApplicationRequest<'_>,
        journal: &AttemptJournalRecord,
    ) -> impl Future<Output = Result<Self::Reply, Self::Error>>;
    /// Decode the exact protocol reply and qualify its original attribution and custody.
    fn qualify(
        &mut self,
        request: &ApplicationRequest<'_>,
        reply: &Self::Reply,
    ) -> impl Future<Output = Result<ApplicationQualification, Self::Error>>;
}

/// Successful local settlement, without terminal, paid-effect or reference-release authority.
#[derive(Debug)]
pub struct ApplicationExecution<R> {
    /// Exact owned response, including original transport evidence.
    pub reply: R,
    /// Explicit qualified outcome; no observation result grants automatic mutation reissue.
    pub outcome: ObservationOutcomeRecord,
    /// Existing checkpoint only when the original mutation is Applied and the stage complete.
    pub retained: Option<RetainedStageEvidence>,
}

/// Exact local or integration refusal, with original spending retained.
#[derive(Debug, thiserror::Error)]
pub enum ApplicationExecutionCause<E: std::error::Error + 'static> {
    /// Original stage/journal/reservation admission failed.
    #[error(transparent)]
    Reservation(#[from] ExecutionReservationError),
    /// Original plan cannot supply the requested operation.
    #[error(transparent)]
    Plan(#[from] OperationPlanError),
    /// Original opaque payload or reserved request differs.
    #[error(transparent)]
    Request(#[from] ApplicationIngressError),
    /// Receipt persistence or current journal admission failed.
    #[error(transparent)]
    Journal(#[from] AttemptJournalError),
    /// Existing stage checkpoint failed after journal exclusion was released.
    #[error(transparent)]
    Checkpoint(#[from] ExecutionStageCheckpointError),
    /// Required application admission, transport or qualification failed.
    #[error("application ingress callback failed: {0}")]
    Application(#[source] E),
    /// This API checkpoints a complete singleton stage.
    #[error("application ingress requires a singleton stage")]
    NonSingletonStage,
    /// No unresolved original mutation exists for observation reservation.
    #[error("original pending mutation is absent")]
    NoPendingMutation,
    /// A direct mutation reply cannot record an inferred negative or uncertain receipt.
    #[error("direct application mutation requires qualified Applied evidence")]
    MutationOutcome,
    /// The provider violated the original reply ceiling.
    #[error("application reply exceeds original ceiling")]
    ReplyTooLarge,
}

/// A returned response survives every post-reply failure; cancellation still needs durable custody.
#[derive(Debug, thiserror::Error)]
pub enum ApplicationExecutionError<E: std::error::Error + 'static, R: std::fmt::Debug> {
    /// No reply was returned; original consumed authority remains occupied.
    #[error(transparent)]
    BeforeReply(#[from] ApplicationExecutionCause<E>),
    /// Exact response remains owned even if qualification, persistence or checkpointing failed.
    #[error("application reply could not settle: {source}")]
    AfterReply {
        /// Typed refusal, never replacement spending authority.
        source: ApplicationExecutionCause<E>,
        /// Original bounded response and transport evidence.
        reply: Box<R>,
    },
}

/// Execute one new original opaque mutation in a singleton stage and publish its existing stage checkpoint.
/// # Errors
/// Rejects changed/occupied originals, admission, submission or unqualified replies.
pub async fn execute<P: ApplicationIngressProvider>(
    stage: &ExecutionStageGuard<'_>,
    sequence: u64,
    payload: &ApplicationIngressPayload,
    provider: &mut P,
) -> Result<ApplicationExecution<P::Reply>, ApplicationExecutionError<P::Error, P::Reply>> {
    run(stage, sequence, payload, provider, false).await
}

/// Observe an original pending mutation using separate finite observation authority.
///
/// Lost observations remain occupied. NotApplied/Uncertain settle only that observation;
/// they publish no complete stage checkpoint and supply no automatic mutation resend.
/// # Errors
/// Rejects absent/changed pending originals, spent observations and unqualified responses.
pub async fn reconcile<P: ApplicationIngressProvider>(
    stage: &ExecutionStageGuard<'_>,
    sequence: u64,
    payload: &ApplicationIngressPayload,
    provider: &mut P,
) -> Result<ApplicationExecution<P::Reply>, ApplicationExecutionError<P::Error, P::Reply>> {
    run(stage, sequence, payload, provider, true).await
}

async fn run<P: ApplicationIngressProvider>(
    stage: &ExecutionStageGuard<'_>,
    sequence: u64,
    payload: &ApplicationIngressPayload,
    provider: &mut P,
    observation: bool,
) -> Result<ApplicationExecution<P::Reply>, ApplicationExecutionError<P::Error, P::Reply>> {
    type Cause<E> = ApplicationExecutionCause<E>;
    if stage.plan().operations().len() != 1 {
        return Err(Cause::NonSingletonStage.into());
    }
    let authority = stage
        .plan()
        .attempt_authority(sequence)
        .map_err(Cause::from)?;
    if observation {
        payload.validate_observation_binding(authority.binding())
    } else {
        payload.validate_mutation_binding(authority.binding())
    }
    .map_err(Cause::from)?;
    let (reply, result) = super::ic_mutation::dispatch_reserved(
        stage,
        &authority,
        provider,
        |journal| {
            if observation {
                let mutation = journal
                    .record()?
                    .view()
                    .pending_mutation
                    .ok_or(Cause::NoPendingMutation)?;
                journal
                    .reserve_planned_observation(
                        &stage.plan().digest(),
                        mutation,
                        payload.digest().hash(),
                    )
                    .map_err(ExecutionReservationError::from)?;
            } else {
                journal
                    .reserve_planned_mutation(&stage.plan().digest())
                    .map_err(ExecutionReservationError::from)?;
            }
            Ok::<_, Cause<P::Error>>(())
        },
        |journal| {
            Ok(if observation {
                ApplicationRequest::Observation(ApplicationObservationRequest::new(
                    stage.plan(),
                    sequence,
                    journal,
                    payload,
                )?)
            } else {
                ApplicationRequest::Mutation(ApplicationMutationRequest::new(
                    stage.plan(),
                    sequence,
                    journal,
                    payload,
                )?)
            })
        },
        async |provider, request, journal| {
            provider.check(request).map_err(Cause::Application)?;
            provider.admit(request).await.map_err(Cause::Application)?;
            stage.layout().map_err(ExecutionReservationError::from)?;
            request.validate(journal)?;
            provider.check(request).map_err(Cause::Application)?;
            provider
                .submit(request, journal)
                .await
                .map_err(Cause::Application)
        },
        async |provider, journal, request, reply| {
            stage.layout().map_err(ExecutionReservationError::from)?;
            request.validate(journal.record()?)?;
            provider.check(request).map_err(Cause::Application)?;
            if reply.bytes().len() > payload.reply_limit() {
                return Err(Cause::ReplyTooLarge);
            }
            let qualified = provider
                .qualify(request, reply)
                .await
                .map_err(Cause::Application)?;
            stage.layout().map_err(ExecutionReservationError::from)?;
            request.validate(journal.record()?)?;
            provider.check(request).map_err(Cause::Application)?;
            request.record_qualified(journal, &qualified)?;
            Ok(qualified)
        },
    )
    .await?;
    // dispatch_reserved has released the attempt guard before checkpoint publication.
    let result = result.and_then(|qualified| qualified.checkpoint(stage).map_err(Cause::from));
    let (reply, (outcome, retained)) =
        super::qualified_receipt::retain_reply(reply, result, |source, reply| {
            ApplicationExecutionError::AfterReply { source, reply }
        })?;
    Ok(ApplicationExecution {
        reply,
        outcome,
        retained,
    })
}

#[cfg(test)]
mod tests;