use crate::{
model::{
ic_mutation::{IcMutationAcknowledgement, IcMutationRequest, IcMutationRequestError},
ic_request::{IcManagementMethodRecord, IcManagementRequestRecord},
operation_plan::OperationPlanError,
},
ops::persistence::{
AttemptJournalError, AttemptJournalGuard, ExecutionProgressPersistenceError,
ExecutionStageGuard, ExecutionWorkflowPersistenceError,
},
policy::ic_mutation::{IcMutationAssociationError, validate_acknowledgement},
ports::ic_mutation::{IcMutationProvider, IcMutationProviderError},
};
use thiserror::Error;
pub async fn capture_snapshot<E: std::error::Error + 'static>(
stage: &ExecutionStageGuard<'_>,
operation_sequence: u64,
payload: &IcManagementRequestRecord,
provider: &mut impl IcMutationProvider,
admit: impl AsyncFnOnce(&IcMutationRequest<'_>) -> Result<(), E>,
) -> Result<IcMutationAcknowledgement, IcSnapshotCaptureExecutionError<E>> {
if payload.method() != IcManagementMethodRecord::TakeCanisterSnapshot {
return Err(IcSnapshotCaptureExecutionError::CaptureRequired);
}
let plan = stage.plan();
let authority = plan.attempt_authority(operation_sequence)?;
payload
.validate_mutation_binding(authority.binding())
.map_err(IcMutationRequestError::from)?;
let mut journal = AttemptJournalGuard::open(stage.layout()?, &authority)?;
journal.reserve_planned_mutation(&plan.digest())?;
let request = IcMutationRequest::new(plan, operation_sequence, journal.record()?, payload)?;
let acknowledgement = super::reserved_effect::submit_once(
&journal,
&request,
async |request| {
admit(request)
.await
.map_err(IcSnapshotCaptureExecutionError::Admission)
},
|| {
stage
.layout()
.map(|_| ())
.map_err(IcSnapshotCaptureExecutionError::Stage)
},
async |request, record| {
provider
.submit_mutation(request, record)
.await
.map_err(IcSnapshotCaptureExecutionError::Provider)
},
IcSnapshotCaptureExecutionError::Journal,
)
.await?;
if let Err(error) = super::reserved_effect::check_returned(
stage,
&journal,
&request,
&acknowledgement,
|request, record, reply| validate_acknowledgement(request, record, reply).map(|_| ()),
) {
return Err(match error {
super::reserved_effect::ReturnedError::Journal(source) => {
IcSnapshotCaptureExecutionError::AfterReplyJournal {
source,
acknowledgement: Box::new(acknowledgement),
}
}
super::reserved_effect::ReturnedError::Association(source) => {
IcSnapshotCaptureExecutionError::Association {
source,
acknowledgement: Box::new(acknowledgement),
}
}
super::reserved_effect::ReturnedError::Stage(source) => {
IcSnapshotCaptureExecutionError::AfterReplyStage {
source,
acknowledgement: Box::new(acknowledgement),
}
}
});
}
Ok(acknowledgement)
}
#[derive(Debug, Error)]
pub enum IcSnapshotCaptureExecutionError<E: std::error::Error + 'static> {
#[error("snapshot capture requires take_canister_snapshot")]
CaptureRequired,
#[error(transparent)]
Plan(#[from] OperationPlanError),
#[error(transparent)]
Stage(#[from] ExecutionWorkflowPersistenceError),
#[error(transparent)]
Journal(#[from] AttemptJournalError),
#[error(transparent)]
Progress(#[from] ExecutionProgressPersistenceError),
#[error(transparent)]
Request(#[from] IcMutationRequestError),
#[error("fresh snapshot capture admission failed: {0}")]
Admission(#[source] E),
#[error(transparent)]
Provider(#[from] IcMutationProviderError),
#[error("snapshot capture acknowledgement association failed: {source}")]
Association {
source: IcMutationAssociationError,
acknowledgement: Box<IcMutationAcknowledgement>,
},
#[error("snapshot capture journal changed after reply: {source}")]
AfterReplyJournal {
source: AttemptJournalError,
acknowledgement: Box<IcMutationAcknowledgement>,
},
#[error("snapshot capture stage changed after reply: {source}")]
AfterReplyStage {
source: ExecutionWorkflowPersistenceError,
acknowledgement: Box<IcMutationAcknowledgement>,
},
}
#[cfg(all(test, unix))]
mod tests;