use crate::{
model::{
artifacts::ArtifactChecksumRecord,
attempt_journal::MutationReceiptRequest,
execution_workflow::ExecutionStagePredecessorRecord,
ic_snapshot_metadata::IcSnapshotMetadataRequest,
ic_snapshot_transfer_read::{
IcSnapshotTransferReadError, IcSnapshotTransferReadPayload,
IcSnapshotTransferReadRequest, IcSnapshotTransferReadResponse,
},
},
ops::persistence::{
AttemptJournalError, AttemptJournalGuard, ExecutionStageCheckpointError,
ExecutionStageGuard, ExecutionWorkflowPersistenceError,
},
policy::ic_snapshot_transfer_read::{
IcSnapshotTransferReadAssociationError, IcSnapshotTransferReadReply,
IcSnapshotTransferReadView,
},
ports::ic_snapshot_transfer_read::IcSnapshotTransferReadProvider,
workflow::ic_snapshot_transfer_read::{
IcSnapshotTransferReadExecutionError, read_snapshot_with,
},
};
use thiserror::Error;
pub async fn read_snapshot_metadata<E: std::error::Error + 'static>(
stage: &ExecutionStageGuard<'_>,
operation_sequence: u64,
payload: &IcSnapshotMetadataRequest,
provider: &mut impl IcSnapshotTransferReadProvider,
admit: impl AsyncFnOnce(&IcSnapshotTransferReadRequest<'_, '_>) -> Result<(), E>,
qualify: impl AsyncFnOnce(
&IcSnapshotTransferReadRequest<'_, '_>,
&IcSnapshotTransferReadResponse,
) -> Result<MutationReceiptRequest, E>,
) -> Result<
(
IcSnapshotTransferReadResponse,
ExecutionStagePredecessorRecord,
),
IcSnapshotMetadataExecutionError<E>,
> {
let operations = stage.plan().operations();
if operations.len() != 1
|| operations[0].operation_sequence() != operation_sequence
|| operations[0].request() != payload.digest().hash()
{
return Err(IcSnapshotMetadataExecutionError::OriginalMismatch);
}
let (response, settlement) = read_snapshot_with(
stage,
operation_sequence,
IcSnapshotTransferReadPayload::Metadata(payload),
provider,
admit,
async |journal, request, admitted| {
settle_metadata(stage, journal, request, admitted, qualify).await
},
)
.await?;
match settlement.and_then(|digest| Ok(stage.reconcile_checkpoint(digest)?)) {
Ok(predecessor) => Ok((response, predecessor)),
Err(source) => Err(IcSnapshotMetadataExecutionError::AfterReply {
source,
response: Box::new(response),
}),
}
}
async fn settle_metadata<E: std::error::Error + 'static>(
stage: &ExecutionStageGuard<'_>,
journal: &mut AttemptJournalGuard<'_>,
request: &IcSnapshotTransferReadRequest<'_, '_>,
admitted: &IcSnapshotTransferReadView<'_, '_>,
qualify: impl AsyncFnOnce(
&IcSnapshotTransferReadRequest<'_, '_>,
&IcSnapshotTransferReadResponse,
) -> Result<MutationReceiptRequest, E>,
) -> Result<ArtifactChecksumRecord, IcSnapshotMetadataSettlementError<E>> {
super::qualified_receipt::AppliedReceiptRecorder {
journal,
expected_attempt: request.mutation_attempt(),
expected_request: request.payload().digest().hash(),
}
.record_applied(
request,
admitted.response(),
async |request, reply| qualify(request, reply).await,
IcSnapshotMetadataSettlementError::Qualification,
|| IcSnapshotMetadataSettlementError::ReceiptRequired,
|| {
let digest = match admitted.reply() {
IcSnapshotTransferReadReply::Metadata(metadata) => metadata.digest(),
IcSnapshotTransferReadReply::Data(_) => {
return Err(IcSnapshotMetadataSettlementError::ReceiptRequired);
}
};
stage.layout()?;
Ok(digest)
},
)
.await
}
#[derive(Debug, Error)]
pub enum IcSnapshotMetadataExecutionError<E: std::error::Error + 'static> {
#[error("snapshot metadata requires an exact singleton original stage")]
OriginalMismatch,
#[error(transparent)]
Read(#[from] IcSnapshotTransferReadExecutionError<E>),
#[error("snapshot metadata reply settlement failed: {source}")]
AfterReply {
source: IcSnapshotMetadataSettlementError<E>,
response: Box<IcSnapshotTransferReadResponse>,
},
}
#[derive(Debug, Error)]
pub enum IcSnapshotMetadataSettlementError<E: std::error::Error + 'static> {
#[error("snapshot metadata qualification failed: {0}")]
Qualification(#[source] E),
#[error("snapshot metadata requires an explicit original Applied receipt")]
ReceiptRequired,
#[error(transparent)]
Stage(#[from] ExecutionWorkflowPersistenceError),
#[error(transparent)]
Journal(#[from] AttemptJournalError),
#[error(transparent)]
Request(#[from] IcSnapshotTransferReadError),
#[error(transparent)]
Association(#[from] IcSnapshotTransferReadAssociationError),
#[error(transparent)]
Checkpoint(#[from] ExecutionStageCheckpointError),
}
#[cfg(all(test, unix))]
mod tests;