ic_backup/workflow/ic_snapshot_metadata/
mod.rs1use crate::{
4 model::{
5 artifacts::ArtifactChecksumRecord,
6 attempt_journal::MutationReceiptRequest,
7 execution_workflow::ExecutionStagePredecessorRecord,
8 ic_snapshot_metadata::IcSnapshotMetadataRequest,
9 ic_snapshot_transfer_read::{
10 IcSnapshotTransferReadPayload, IcSnapshotTransferReadRequest,
11 IcSnapshotTransferReadResponse,
12 },
13 },
14 ops::persistence::{
15 AttemptJournalError, AttemptJournalGuard, ExecutionStageCheckpointError,
16 ExecutionStageGuard, ExecutionWorkflowPersistenceError,
17 },
18 policy::ic_snapshot_transfer_read::{IcSnapshotTransferReadReply, IcSnapshotTransferReadView},
19 ports::ic_snapshot_transfer_read::IcSnapshotTransferReadProvider,
20 workflow::ic_snapshot_transfer_read::{
21 IcSnapshotTransferReadExecutionError, read_snapshot_with,
22 },
23};
24use thiserror::Error;
25
26pub async fn read_snapshot_metadata<E: std::error::Error + 'static>(
57 stage: &ExecutionStageGuard<'_>,
58 operation_sequence: u64,
59 payload: &IcSnapshotMetadataRequest,
60 provider: &mut impl IcSnapshotTransferReadProvider,
61 admit: impl AsyncFnOnce(&IcSnapshotTransferReadRequest<'_, '_>) -> Result<(), E>,
62 qualify: impl AsyncFnOnce(
63 &IcSnapshotTransferReadRequest<'_, '_>,
64 &IcSnapshotTransferReadResponse,
65 ) -> Result<MutationReceiptRequest, E>,
66) -> Result<
67 (
68 IcSnapshotTransferReadResponse,
69 ExecutionStagePredecessorRecord,
70 ),
71 IcSnapshotMetadataExecutionError<E>,
72> {
73 let operations = stage.plan().operations();
74 if operations.len() != 1
75 || operations[0].operation_sequence() != operation_sequence
76 || operations[0].request() != payload.digest().hash()
77 {
78 return Err(IcSnapshotMetadataExecutionError::OriginalMismatch);
79 }
80 let (response, settlement) = read_snapshot_with(
81 stage,
82 operation_sequence,
83 IcSnapshotTransferReadPayload::Metadata(payload),
84 provider,
85 admit,
86 async |journal, request, admitted| {
87 settle_metadata(stage, journal, request, admitted, qualify).await
88 },
89 )
90 .await?;
91 super::qualified_receipt::checkpoint_reply(stage, response, settlement, |source, response| {
92 IcSnapshotMetadataExecutionError::AfterReply { source, response }
93 })
94}
95
96async fn settle_metadata<E: std::error::Error + 'static>(
97 stage: &ExecutionStageGuard<'_>,
98 journal: &mut AttemptJournalGuard<'_>,
99 request: &IcSnapshotTransferReadRequest<'_, '_>,
100 admitted: &IcSnapshotTransferReadView<'_, '_>,
101 qualify: impl AsyncFnOnce(
102 &IcSnapshotTransferReadRequest<'_, '_>,
103 &IcSnapshotTransferReadResponse,
104 ) -> Result<MutationReceiptRequest, E>,
105) -> Result<ArtifactChecksumRecord, IcSnapshotMetadataSettlementError<E>> {
106 let receipt = qualify(request, admitted.response())
107 .await
108 .map_err(IcSnapshotMetadataSettlementError::Qualification)?;
109 let receipt = super::qualified_receipt::admit_applied(
110 receipt,
111 request.mutation_attempt(),
112 request.payload().digest().hash(),
113 )
114 .ok_or(IcSnapshotMetadataSettlementError::ReceiptRequired)?;
115 let digest = match admitted.reply() {
116 IcSnapshotTransferReadReply::Metadata(metadata) => metadata.digest(),
117 IcSnapshotTransferReadReply::Data(_) => {
118 return Err(IcSnapshotMetadataSettlementError::ReceiptRequired);
119 }
120 };
121 stage.layout()?;
122 journal.record_mutation(receipt)?;
123 Ok(digest)
124}
125
126#[derive(Debug, Error)]
128pub enum IcSnapshotMetadataExecutionError<E: std::error::Error + 'static> {
129 #[error("snapshot metadata requires an exact singleton original stage")]
131 OriginalMismatch,
132 #[error(transparent)]
134 Read(#[from] IcSnapshotTransferReadExecutionError<E>),
135 #[error("snapshot metadata reply settlement failed: {source}")]
137 AfterReply {
138 source: IcSnapshotMetadataSettlementError<E>,
140 response: Box<IcSnapshotTransferReadResponse>,
142 },
143}
144
145#[derive(Debug, Error)]
147pub enum IcSnapshotMetadataSettlementError<E: std::error::Error + 'static> {
148 #[error("snapshot metadata qualification failed: {0}")]
150 Qualification(#[source] E),
151 #[error("snapshot metadata requires an explicit original Applied receipt")]
153 ReceiptRequired,
154 #[error(transparent)]
156 Stage(#[from] ExecutionWorkflowPersistenceError),
157 #[error(transparent)]
159 Journal(#[from] AttemptJournalError),
160 #[error(transparent)]
162 Checkpoint(#[from] ExecutionStageCheckpointError),
163}
164
165#[cfg(all(test, unix))]
166mod tests;