ic_backup/workflow/ic_snapshot_upload/allocation/
mod.rs1use super::{IcSnapshotUploadExecutionError, upload_snapshot};
4use crate::{
5 model::{
6 attempt_journal::{MutationOutcomeRecord, MutationReceiptRequest},
7 execution_workflow::ExecutionStagePredecessorRecord,
8 ic_mutation::IcMutationAcknowledgement,
9 ic_snapshot_upload::{
10 IcSnapshotUploadAttempt, IcSnapshotUploadAttemptError, IcSnapshotUploadKind,
11 IcSnapshotUploadRequest,
12 },
13 },
14 ops::persistence::{
15 AttemptJournalError, AttemptJournalGuard, ExecutionStageCheckpointError,
16 ExecutionStageGuard, ExecutionWorkflowPersistenceError,
17 },
18 policy::ic_snapshot_upload::{IcSnapshotUploadAssociationError, validate_acknowledgement},
19 ports::ic_snapshot_upload::IcSnapshotUploadProvider,
20};
21use thiserror::Error;
22
23pub async fn allocate_snapshot<E: std::error::Error + 'static>(
52 stage: &ExecutionStageGuard<'_>,
53 operation_sequence: u64,
54 payload: &IcSnapshotUploadRequest<'_>,
55 provider: &mut impl IcSnapshotUploadProvider,
56 admit: impl AsyncFnOnce(&IcSnapshotUploadAttempt<'_, '_>) -> Result<(), E>,
57 qualify: impl AsyncFnOnce(
58 &IcSnapshotUploadAttempt<'_, '_>,
59 &IcMutationAcknowledgement,
60 ) -> Result<MutationReceiptRequest, E>,
61) -> Result<
62 (IcMutationAcknowledgement, ExecutionStagePredecessorRecord),
63 IcSnapshotAllocationExecutionError<E>,
64> {
65 let operations = stage.plan().operations();
66 if !matches!(payload.kind(), IcSnapshotUploadKind::Metadata)
67 || operations.len() != 1
68 || operations[0].operation_sequence() != operation_sequence
69 || operations[0].request() != payload.binding_digest().hash()
70 {
71 return Err(IcSnapshotAllocationExecutionError::OriginalMismatch);
72 }
73 let acknowledgement =
74 upload_snapshot(stage, operation_sequence, payload, provider, admit).await?;
75 match settle_allocation(
76 stage,
77 operation_sequence,
78 payload,
79 &acknowledgement,
80 qualify,
81 )
82 .await
83 {
84 Ok(predecessor) => Ok((acknowledgement, predecessor)),
85 Err(source) => Err(IcSnapshotAllocationExecutionError::AfterReply {
86 source,
87 acknowledgement: Box::new(acknowledgement),
88 }),
89 }
90}
91
92async fn settle_allocation<E: std::error::Error + 'static>(
93 stage: &ExecutionStageGuard<'_>,
94 sequence: u64,
95 payload: &IcSnapshotUploadRequest<'_>,
96 acknowledgement: &IcMutationAcknowledgement,
97 qualify: impl AsyncFnOnce(
98 &IcSnapshotUploadAttempt<'_, '_>,
99 &IcMutationAcknowledgement,
100 ) -> Result<MutationReceiptRequest, E>,
101) -> Result<ExecutionStagePredecessorRecord, IcSnapshotAllocationSettlementError<E>> {
102 let plan = stage.plan();
103 let authority = plan
104 .attempt_authority(sequence)
105 .map_err(IcSnapshotUploadAttemptError::from)?;
106 let mut journal = AttemptJournalGuard::open(stage.layout()?, &authority)?;
107 let request = IcSnapshotUploadAttempt::new(plan, sequence, journal.record()?, payload)?;
108 let admitted = validate_acknowledgement(&request, journal.record()?, acknowledgement)?;
109 let receipt = qualify(&request, acknowledgement)
110 .await
111 .map_err(IcSnapshotAllocationSettlementError::Qualification)?;
112 if receipt.outcome != MutationOutcomeRecord::Applied
113 || receipt.attempt != request.mutation_attempt()
114 || receipt.request != payload.binding_digest().hash()
115 {
116 return Err(IcSnapshotAllocationSettlementError::ReceiptRequired);
117 }
118 stage.layout()?;
119 journal.record_mutation(receipt)?;
120 drop(journal);
121 Ok(stage.checkpoint(admitted.reply().digest())?)
122}
123
124#[derive(Debug, Error)]
126pub enum IcSnapshotAllocationExecutionError<E: std::error::Error + 'static> {
127 #[error("snapshot allocation requires an exact singleton original metadata stage")]
129 OriginalMismatch,
130 #[error(transparent)]
132 Upload(#[from] IcSnapshotUploadExecutionError<E>),
133 #[error("snapshot allocation reply settlement failed: {source}")]
135 AfterReply {
136 source: IcSnapshotAllocationSettlementError<E>,
138 acknowledgement: Box<IcMutationAcknowledgement>,
140 },
141}
142
143#[derive(Debug, Error)]
145pub enum IcSnapshotAllocationSettlementError<E: std::error::Error + 'static> {
146 #[error("snapshot allocation qualification failed: {0}")]
148 Qualification(#[source] E),
149 #[error("snapshot allocation requires an explicit original Applied receipt")]
151 ReceiptRequired,
152 #[error(transparent)]
154 Stage(#[from] ExecutionWorkflowPersistenceError),
155 #[error(transparent)]
157 Journal(#[from] AttemptJournalError),
158 #[error(transparent)]
160 Request(#[from] IcSnapshotUploadAttemptError),
161 #[error(transparent)]
163 Association(#[from] IcSnapshotUploadAssociationError),
164 #[error(transparent)]
166 Checkpoint(#[from] ExecutionStageCheckpointError),
167}
168
169#[cfg(all(test, unix))]
170mod tests;