ic-backup 0.64.0

//! Single original mutation reservation and dispatch under continuous journal exclusion.

use crate::ports::IcProviderError;
use crate::{
    model::{
        attempt_journal::AttemptAuthorityRecord,
        ic_mutation::{IcMutationAcknowledgement, IcMutationRequest, IcMutationRequestError},
        ic_request::IcManagementRequestRecord,
    },
    ops::persistence::{
        AttemptJournalError, AttemptJournalGuard, ExecutionProgressPersistenceError,
        ExecutionStageGuard, ExecutionWorkflowPersistenceError,
    },
    ports::ic_mutation::IcMutationProvider,
};

// Domain callbacks retain their admission/re-admission order and post-reply errors.
// The journal drops on return, before a caller can publish a stage checkpoint.
pub(super) async fn submit_with<E, T>(
    stage: &ExecutionStageGuard<'_>,
    authority: &AttemptAuthorityRecord,
    payload: &IcManagementRequestRecord,
    provider: &mut impl IcMutationProvider,
    admit: impl AsyncFnOnce(&IcMutationRequest<'_>) -> Result<(), E>,
    complete: impl AsyncFnOnce(
        &mut AttemptJournalGuard<'_>,
        &IcMutationRequest<'_>,
        &IcMutationAcknowledgement,
    ) -> T,
) -> Result<(IcMutationAcknowledgement, T), E>
where
    E: From<IcMutationRequestError> + From<ExecutionReservationError> + From<IcProviderError>,
{
    payload
        .validate_mutation_binding(authority.binding())
        .map_err(IcMutationRequestError::from)
        .map_err(E::from)?;
    execute_reserved(
        stage,
        authority,
        |_| Ok(()),
        |journal| {
            IcMutationRequest::new(
                stage.plan(),
                authority.binding().operation_sequence(),
                journal,
                payload,
            )
            .map_err(E::from)
        },
        admit,
        async |request, journal| {
            provider
                .submit_mutation(request, journal)
                .await
                .map_err(E::from)
        },
        complete,
    )
    .await
}

// The sole open -> original validation -> reserve -> bind -> admit -> submit
// sequence. Request/reply semantics stay with each typed caller. Keep the same
// journal through both awaits and completion; checkpointing happens after return.
pub(super) async fn execute_reserved<Q, R, T, E>(
    stage: &ExecutionStageGuard<'_>,
    authority: &AttemptAuthorityRecord,
    validate: impl FnOnce(&crate::model::attempt_journal::AttemptJournalRecord) -> Result<(), E>,
    bind: impl FnOnce(&crate::model::attempt_journal::AttemptJournalRecord) -> Result<Q, E>,
    admit: impl AsyncFnOnce(&Q) -> Result<(), E>,
    submit: impl AsyncFnOnce(&Q, &crate::model::attempt_journal::AttemptJournalRecord) -> Result<R, E>,
    complete: impl AsyncFnOnce(&mut AttemptJournalGuard<'_>, &Q, &R) -> T,
) -> Result<(R, T), E>
where
    E: From<ExecutionReservationError>,
{
    dispatch_reserved(
        stage,
        authority,
        &mut (),
        |journal| {
            validate(
                journal
                    .record()
                    .map_err(ExecutionReservationError::from)
                    .map_err(E::from)?,
            )?;
            journal
                .reserve_planned_mutation(&stage.plan().digest())
                .map_err(ExecutionReservationError::from)
                .map_err(E::from)?;
            Ok(())
        },
        bind,
        async |(), request, journal| {
            admit(request).await?;
            submit(request, journal).await
        },
        async |(), journal, request, reply| complete(journal, request, reply).await,
    )
    .await
}

// One exclusion/binding/dispatch/completion owner for management and opaque ingress.
// Reservation selects the existing mutation or observation journal transition.
pub(super) async fn dispatch_reserved<Q, R, T, E, C>(
    stage: &ExecutionStageGuard<'_>,
    authority: &AttemptAuthorityRecord,
    context: &mut C,
    reserve: impl FnOnce(&mut AttemptJournalGuard<'_>) -> Result<(), E>,
    bind: impl FnOnce(&crate::model::attempt_journal::AttemptJournalRecord) -> Result<Q, E>,
    dispatch: impl AsyncFnOnce(
        &mut C,
        &Q,
        &crate::model::attempt_journal::AttemptJournalRecord,
    ) -> Result<R, E>,
    complete: impl AsyncFnOnce(&mut C, &mut AttemptJournalGuard<'_>, &Q, &R) -> T,
) -> Result<(R, T), E>
where
    E: From<ExecutionReservationError>,
{
    let mut journal = AttemptJournalGuard::open(
        stage
            .layout()
            .map_err(ExecutionReservationError::from)
            .map_err(E::from)?,
        authority,
    )
    .map_err(ExecutionReservationError::from)
    .map_err(E::from)?;
    reserve(&mut journal)?;
    let request = bind(
        journal
            .record()
            .map_err(ExecutionReservationError::from)
            .map_err(E::from)?,
    )?;
    let reply = dispatch(
        context,
        &request,
        journal
            .record()
            .map_err(ExecutionReservationError::from)
            .map_err(E::from)?,
    )
    .await?;
    let result = complete(context, &mut journal, &request, &reply).await;
    Ok((reply, result))
}

/// Common original-stage and durable-reservation failure before a reply exists.
/// Every failure retains original spending and supplies no retry authority.
#[derive(Debug, thiserror::Error)]
pub enum ExecutionReservationError {
    /// Retained stage or ancestor admission failed.
    #[error(transparent)]
    Stage(#[from] ExecutionWorkflowPersistenceError),
    /// Original selected journal could not be admitted.
    #[error(transparent)]
    Journal(#[from] AttemptJournalError),
    /// Complete original progress or bounded reservation failed.
    #[error(transparent)]
    Progress(#[from] ExecutionProgressPersistenceError),
}

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