ic_backup/workflow/ic_snapshot_transfer_read/
mod.rs1use crate::{
4 model::{
5 ic_snapshot_transfer_read::{
6 IcSnapshotTransferReadError, IcSnapshotTransferReadPayload,
7 IcSnapshotTransferReadRequest, IcSnapshotTransferReadResponse,
8 },
9 operation_plan::OperationPlanError,
10 },
11 ops::persistence::{
12 AttemptJournalError, AttemptJournalGuard, ExecutionProgressPersistenceError,
13 ExecutionStageGuard, ExecutionWorkflowPersistenceError,
14 },
15 policy::ic_snapshot_transfer_read::{
16 IcSnapshotTransferReadAssociationError, IcSnapshotTransferReadView, validate_response,
17 },
18 ports::{
19 ic_observation::IcObservationProviderError,
20 ic_snapshot_transfer_read::IcSnapshotTransferReadProvider,
21 },
22};
23use thiserror::Error;
24
25pub async fn read_snapshot<E: std::error::Error + 'static>(
50 stage: &ExecutionStageGuard<'_>,
51 operation_sequence: u64,
52 payload: IcSnapshotTransferReadPayload<'_, '_>,
53 provider: &mut impl IcSnapshotTransferReadProvider,
54 admit: impl AsyncFnOnce(&IcSnapshotTransferReadRequest<'_, '_>) -> Result<(), E>,
55) -> Result<IcSnapshotTransferReadResponse, IcSnapshotTransferReadExecutionError<E>> {
56 let (response, ()) = read_snapshot_with(
57 stage,
58 operation_sequence,
59 payload,
60 provider,
61 admit,
62 async |_, _, _| (),
63 )
64 .await?;
65 Ok(response)
66}
67
68pub(super) async fn read_snapshot_with<E: std::error::Error + 'static, T>(
72 stage: &ExecutionStageGuard<'_>,
73 operation_sequence: u64,
74 payload: IcSnapshotTransferReadPayload<'_, '_>,
75 provider: &mut impl IcSnapshotTransferReadProvider,
76 admit: impl AsyncFnOnce(&IcSnapshotTransferReadRequest<'_, '_>) -> Result<(), E>,
77 complete: impl AsyncFnOnce(
78 &mut AttemptJournalGuard<'_>,
79 &IcSnapshotTransferReadRequest<'_, '_>,
80 &IcSnapshotTransferReadView<'_, '_>,
81 ) -> T,
82) -> Result<(IcSnapshotTransferReadResponse, T), IcSnapshotTransferReadExecutionError<E>> {
83 let plan = stage.plan();
84 let authority = plan.attempt_authority(operation_sequence)?;
85 payload.validate_binding(&authority)?;
86 let mut journal = AttemptJournalGuard::open(stage.layout()?, &authority)?;
87 journal.reserve_planned_mutation(&plan.digest())?;
88 let request =
89 IcSnapshotTransferReadRequest::new(plan, operation_sequence, journal.record()?, payload)?;
90 admit(&request)
91 .await
92 .map_err(IcSnapshotTransferReadExecutionError::Admission)?;
93 stage.layout()?;
94 let response = provider.read_snapshot(&request, journal.record()?).await?;
95 let admit_reply = || {
96 let admitted = validate_response(&request, journal.record()?, &response)?;
97 stage.layout()?;
98 Ok::<_, IcSnapshotTransferReadReplyError>(admitted)
99 };
100 let admitted = match admit_reply() {
101 Ok(admitted) => admitted,
102 Err(source) => {
103 return Err(IcSnapshotTransferReadExecutionError::AfterReply {
104 source,
105 response: Box::new(response),
106 });
107 }
108 };
109 let result = complete(&mut journal, &request, &admitted).await;
110 Ok((response, result))
111}
112
113#[derive(Debug, Error)]
115pub enum IcSnapshotTransferReadExecutionError<E: std::error::Error + 'static> {
116 #[error(transparent)]
118 Plan(#[from] OperationPlanError),
119 #[error(transparent)]
121 Stage(#[from] ExecutionWorkflowPersistenceError),
122 #[error(transparent)]
124 Journal(#[from] AttemptJournalError),
125 #[error(transparent)]
127 Progress(#[from] ExecutionProgressPersistenceError),
128 #[error(transparent)]
130 Request(#[from] IcSnapshotTransferReadError),
131 #[error("fresh snapshot read admission failed: {0}")]
133 Admission(#[source] E),
134 #[error(transparent)]
136 Provider(#[from] IcObservationProviderError),
137 #[error("snapshot transfer read reply rejected: {source}")]
139 AfterReply {
140 source: IcSnapshotTransferReadReplyError,
142 response: Box<IcSnapshotTransferReadResponse>,
144 },
145}
146
147#[derive(Debug, Error)]
149pub enum IcSnapshotTransferReadReplyError {
150 #[error(transparent)]
152 Journal(#[from] AttemptJournalError),
153 #[error(transparent)]
155 Association(#[from] IcSnapshotTransferReadAssociationError),
156 #[error(transparent)]
158 Stage(#[from] ExecutionWorkflowPersistenceError),
159}
160
161#[cfg(all(test, unix))]
162mod tests;