mod source;
pub use source::{IcRestoreSource, IcRestoreSourceError};
use crate::{
model::{
attempt_journal::MutationReceiptRequest,
execution_workflow::ExecutionStagePredecessorRecord,
ic_mutation::{IcMutationAcknowledgement, IcMutationRequest, IcMutationRequestError},
restore_safety::{RestoreSafetyObservation, RestoreSafetyRequest},
},
ops::persistence::{
AttemptJournalError, ExecutionProgressPersistenceError, ExecutionStageCheckpointError,
ExecutionStageGuard, ExecutionWorkflowPersistenceError, RestoreSafetyPersistenceError,
read_restore_safety_requirement,
},
policy::{
ic_mutation::{IcMutationAssociationError, IcMutationReplyView, validate_acknowledgement},
restore_safety::{RestoreSafetyError, validate},
},
ports::ic_mutation::{IcMutationProvider, IcMutationProviderError},
};
use thiserror::Error;
pub async fn restore_snapshot<E: std::error::Error + 'static>(
stage: &ExecutionStageGuard<'_>,
source: IcRestoreSource<'_>,
safety: &RestoreSafetyRequest<'_>,
provider: &mut impl IcMutationProvider,
admit: impl AsyncFnOnce(
&IcMutationRequest<'_>,
&RestoreSafetyRequest<'_>,
) -> Result<RestoreSafetyObservation, E>,
qualify: impl AsyncFnOnce(
&IcMutationRequest<'_>,
&IcMutationAcknowledgement,
) -> Result<MutationReceiptRequest, E>,
) -> Result<(IcMutationAcknowledgement, ExecutionStagePredecessorRecord), IcRestoreExecutionError<E>>
{
let plan = stage.plan();
let sequence = safety.binding().operation_sequence();
if plan.operations().len() != 1 || plan.operations()[0].operation_sequence() != sequence {
return Err(IcRestoreExecutionError::OriginalMismatch);
}
let retain = || {
source.validate(stage, safety.wire())?;
read_restore_safety_requirement(
stage.layout()?,
source.layout,
plan,
source.plan,
&safety.requirement().digest(),
)?;
Ok::<_, IcRestoreReplyError<E>>(())
};
retain()?;
let authority = plan
.attempt_authority(sequence)
.map_err(IcMutationRequestError::from)?;
let (acknowledgement, settlement) = super::ic_mutation::submit_with(
stage,
&authority,
safety.wire(),
provider,
async |request| {
let observation = admit(request, safety)
.await
.map_err(IcRestoreExecutionError::Admission)?;
validate(safety, &observation)?;
retain()?;
Ok::<_, IcRestoreExecutionError<E>>(())
},
async |journal, request, acknowledgement| {
let admitted = validate_acknowledgement(request, journal.record()?, acknowledgement)?;
retain()?;
let receipt = qualify(request, acknowledgement)
.await
.map_err(IcRestoreReplyError::Qualification)?;
let receipt = super::qualified_receipt::admit_applied(
receipt,
request.mutation_attempt(),
safety.wire().digest().hash(),
)
.ok_or(IcRestoreReplyError::ReceiptRequired)?;
retain()?;
journal.record_mutation(receipt)?;
retain()?;
let IcMutationReplyView::Lifecycle(reply) = admitted.reply() else {
return Err(IcRestoreReplyError::ReceiptRequired);
};
Ok::<_, IcRestoreReplyError<E>>(reply.digest())
},
)
.await?;
super::qualified_receipt::checkpoint_reply(
stage,
acknowledgement,
settlement,
|source, acknowledgement| IcRestoreExecutionError::AfterReply {
source,
acknowledgement,
},
)
}
#[derive(Debug, Error)]
pub enum IcRestoreExecutionError<E: std::error::Error + 'static> {
#[error("restore requires an exact singleton original load/start stage")]
OriginalMismatch,
#[error(transparent)]
Stage(#[from] ExecutionWorkflowPersistenceError),
#[error(transparent)]
Journal(#[from] AttemptJournalError),
#[error(transparent)]
Progress(#[from] ExecutionProgressPersistenceError),
#[error(transparent)]
Request(#[from] IcMutationRequestError),
#[error(transparent)]
Retention(#[from] IcRestoreReplyError<E>),
#[error("fresh restore admission failed: {0}")]
Admission(#[source] E),
#[error(transparent)]
Safety(#[from] RestoreSafetyError),
#[error(transparent)]
Provider(#[from] IcMutationProviderError),
#[error("restore reply settlement failed: {source}")]
AfterReply {
source: IcRestoreReplyError<E>,
acknowledgement: Box<IcMutationAcknowledgement>,
},
}
#[derive(Debug, Error)]
pub enum IcRestoreReplyError<E: std::error::Error + 'static> {
#[error(transparent)]
Source(#[from] IcRestoreSourceError),
#[error(transparent)]
Stage(#[from] ExecutionWorkflowPersistenceError),
#[error(transparent)]
Retention(#[from] RestoreSafetyPersistenceError),
#[error(transparent)]
Journal(#[from] AttemptJournalError),
#[error(transparent)]
Association(#[from] IcMutationAssociationError),
#[error("restore reply qualification failed: {0}")]
Qualification(#[source] E),
#[error("restore requires an explicit original Applied receipt")]
ReceiptRequired,
#[error(transparent)]
Checkpoint(#[from] ExecutionStageCheckpointError),
}
#[cfg(test)]
mod tests;