ic_backup/workflow/ic_snapshot_metadata/
mod.rs1use crate::{
4 model::{
5 attempt_journal::MutationReceiptRequest,
6 execution_workflow::ExecutionStagePredecessorRecord,
7 ic_snapshot_metadata::IcSnapshotMetadataRequest,
8 ic_snapshot_transfer_read::{
9 IcSnapshotTransferReadError, IcSnapshotTransferReadPayload,
10 IcSnapshotTransferReadRequest, IcSnapshotTransferReadResponse,
11 },
12 },
13 ops::persistence::{
14 AttemptJournalError, AttemptJournalGuard, ExecutionStageCheckpointError,
15 ExecutionStageGuard, ExecutionWorkflowPersistenceError,
16 },
17 policy::ic_snapshot_transfer_read::{
18 IcSnapshotTransferReadAssociationError, IcSnapshotTransferReadReply, validate_response,
19 },
20 ports::ic_snapshot_transfer_read::IcSnapshotTransferReadProvider,
21 workflow::ic_snapshot_transfer_read::{IcSnapshotTransferReadExecutionError, read_snapshot},
22};
23use thiserror::Error;
24
25pub async fn read_snapshot_metadata<E: std::error::Error + 'static>(
55 stage: &ExecutionStageGuard<'_>,
56 operation_sequence: u64,
57 payload: &IcSnapshotMetadataRequest,
58 provider: &mut impl IcSnapshotTransferReadProvider,
59 admit: impl AsyncFnOnce(&IcSnapshotTransferReadRequest<'_, '_>) -> Result<(), E>,
60 qualify: impl AsyncFnOnce(
61 &IcSnapshotTransferReadRequest<'_, '_>,
62 &IcSnapshotTransferReadResponse,
63 ) -> Result<MutationReceiptRequest, E>,
64) -> Result<
65 (
66 IcSnapshotTransferReadResponse,
67 ExecutionStagePredecessorRecord,
68 ),
69 IcSnapshotMetadataExecutionError<E>,
70> {
71 let operations = stage.plan().operations();
72 if operations.len() != 1
73 || operations[0].operation_sequence() != operation_sequence
74 || operations[0].request() != payload.digest().hash()
75 {
76 return Err(IcSnapshotMetadataExecutionError::OriginalMismatch);
77 }
78 let response = read_snapshot(
79 stage,
80 operation_sequence,
81 IcSnapshotTransferReadPayload::Metadata(payload),
82 provider,
83 admit,
84 )
85 .await?;
86 match settle_metadata(stage, operation_sequence, payload, &response, qualify).await {
87 Ok(predecessor) => Ok((response, predecessor)),
88 Err(source) => Err(IcSnapshotMetadataExecutionError::AfterReply {
89 source,
90 response: Box::new(response),
91 }),
92 }
93}
94
95async fn settle_metadata<E: std::error::Error + 'static>(
96 stage: &ExecutionStageGuard<'_>,
97 sequence: u64,
98 payload: &IcSnapshotMetadataRequest,
99 response: &IcSnapshotTransferReadResponse,
100 qualify: impl AsyncFnOnce(
101 &IcSnapshotTransferReadRequest<'_, '_>,
102 &IcSnapshotTransferReadResponse,
103 ) -> Result<MutationReceiptRequest, E>,
104) -> Result<ExecutionStagePredecessorRecord, IcSnapshotMetadataSettlementError<E>> {
105 let plan = stage.plan();
106 let authority = plan
107 .attempt_authority(sequence)
108 .map_err(IcSnapshotTransferReadError::from)?;
109 let mut journal = AttemptJournalGuard::open(stage.layout()?, &authority)?;
110 let request = IcSnapshotTransferReadRequest::new(
111 plan,
112 sequence,
113 journal.record()?,
114 IcSnapshotTransferReadPayload::Metadata(payload),
115 )?;
116 let admitted = validate_response(&request, journal.record()?, response)?;
117 let metadata_digest = super::qualified_receipt::AppliedReceiptRecorder {
118 journal: &mut journal,
119 expected_attempt: request.mutation_attempt(),
120 expected_request: payload.digest().hash(),
121 }
122 .record_applied(
123 &request,
124 response,
125 async |request, reply| qualify(request, reply).await,
126 IcSnapshotMetadataSettlementError::Qualification,
127 || IcSnapshotMetadataSettlementError::ReceiptRequired,
128 || {
129 let digest = match admitted.reply() {
130 IcSnapshotTransferReadReply::Metadata(metadata) => metadata.digest(),
131 IcSnapshotTransferReadReply::Data(_) => {
132 return Err(IcSnapshotMetadataSettlementError::ReceiptRequired);
133 }
134 };
135 stage.layout()?;
136 Ok(digest)
137 },
138 )
139 .await?;
140 drop(journal);
141 Ok(stage.checkpoint(metadata_digest)?)
142}
143
144#[derive(Debug, Error)]
146pub enum IcSnapshotMetadataExecutionError<E: std::error::Error + 'static> {
147 #[error("snapshot metadata requires an exact singleton original stage")]
149 OriginalMismatch,
150 #[error(transparent)]
152 Read(#[from] IcSnapshotTransferReadExecutionError<E>),
153 #[error("snapshot metadata reply settlement failed: {source}")]
155 AfterReply {
156 source: IcSnapshotMetadataSettlementError<E>,
158 response: Box<IcSnapshotTransferReadResponse>,
160 },
161}
162
163#[derive(Debug, Error)]
165pub enum IcSnapshotMetadataSettlementError<E: std::error::Error + 'static> {
166 #[error("snapshot metadata qualification failed: {0}")]
168 Qualification(#[source] E),
169 #[error("snapshot metadata requires an explicit original Applied receipt")]
171 ReceiptRequired,
172 #[error(transparent)]
174 Stage(#[from] ExecutionWorkflowPersistenceError),
175 #[error(transparent)]
177 Journal(#[from] AttemptJournalError),
178 #[error(transparent)]
180 Request(#[from] IcSnapshotTransferReadError),
181 #[error(transparent)]
183 Association(#[from] IcSnapshotTransferReadAssociationError),
184 #[error(transparent)]
186 Checkpoint(#[from] ExecutionStageCheckpointError),
187}
188
189#[cfg(all(test, unix))]
190mod tests;