use super::{IcSnapshotUploadExecutionError, upload_snapshot_with};
use crate::{
model::{
attempt_journal::MutationReceiptRequest,
execution_workflow::ExecutionStagePredecessorRecord,
ic_mutation::IcMutationAcknowledgement,
ic_snapshot_upload::{
IcSnapshotUploadAttempt, IcSnapshotUploadKind, IcSnapshotUploadRequest,
},
},
ops::persistence::{
AttemptJournalError, ExecutionStageCheckpointError, ExecutionStageGuard,
ExecutionWorkflowPersistenceError,
},
ports::ic_snapshot_upload::IcSnapshotUploadProvider,
};
use thiserror::Error;
pub async fn allocate_snapshot<E: std::error::Error + 'static>(
stage: &ExecutionStageGuard<'_>,
operation_sequence: u64,
payload: &IcSnapshotUploadRequest<'_>,
provider: &mut impl IcSnapshotUploadProvider,
admit: impl AsyncFnOnce(&IcSnapshotUploadAttempt<'_, '_>) -> Result<(), E>,
qualify: impl AsyncFnOnce(
&IcSnapshotUploadAttempt<'_, '_>,
&IcMutationAcknowledgement,
) -> Result<MutationReceiptRequest, E>,
) -> Result<
(IcMutationAcknowledgement, ExecutionStagePredecessorRecord),
IcSnapshotAllocationExecutionError<E>,
> {
let operations = stage.plan().operations();
if !matches!(payload.kind(), IcSnapshotUploadKind::Metadata)
|| operations.len() != 1
|| operations[0].operation_sequence() != operation_sequence
|| operations[0].request() != payload.binding_digest().hash()
{
return Err(IcSnapshotAllocationExecutionError::OriginalMismatch);
}
let (acknowledgement, settlement) = upload_snapshot_with(
stage,
operation_sequence,
payload,
provider,
admit,
async |journal, request, admitted| {
let receipt = qualify(request, admitted.acknowledgement())
.await
.map_err(IcSnapshotAllocationSettlementError::Qualification)?;
let receipt = super::super::qualified_receipt::admit_applied(
receipt,
request.mutation_attempt(),
payload.binding_digest().hash(),
)
.ok_or(IcSnapshotAllocationSettlementError::ReceiptRequired)?;
stage.layout()?;
journal.record_mutation(receipt)?;
Ok::<_, IcSnapshotAllocationSettlementError<E>>(admitted.reply().digest())
},
)
.await?;
match settlement.and_then(|evidence| Ok(stage.reconcile_checkpoint(evidence)?)) {
Ok(predecessor) => Ok((acknowledgement, predecessor)),
Err(source) => Err(IcSnapshotAllocationExecutionError::AfterReply {
source,
acknowledgement: Box::new(acknowledgement),
}),
}
}
#[derive(Debug, Error)]
pub enum IcSnapshotAllocationExecutionError<E: std::error::Error + 'static> {
#[error("snapshot allocation requires an exact singleton original metadata stage")]
OriginalMismatch,
#[error(transparent)]
Upload(#[from] IcSnapshotUploadExecutionError<E>),
#[error("snapshot allocation reply settlement failed: {source}")]
AfterReply {
source: IcSnapshotAllocationSettlementError<E>,
acknowledgement: Box<IcMutationAcknowledgement>,
},
}
#[derive(Debug, Error)]
pub enum IcSnapshotAllocationSettlementError<E: std::error::Error + 'static> {
#[error("snapshot allocation qualification failed: {0}")]
Qualification(#[source] E),
#[error("snapshot allocation requires an explicit original Applied receipt")]
ReceiptRequired,
#[error(transparent)]
Stage(#[from] ExecutionWorkflowPersistenceError),
#[error(transparent)]
Journal(#[from] AttemptJournalError),
#[error(transparent)]
Checkpoint(#[from] ExecutionStageCheckpointError),
}
#[cfg(all(test, unix))]
mod tests;