ic_backup/workflow/ic_snapshot_upload/
mod.rs1mod allocation;
4#[cfg(unix)]
5mod data;
6#[cfg(unix)]
7pub use data::{
8 IcSnapshotDataUploadExecutionError, IcSnapshotDataUploadReplyError, upload_snapshot_data,
9};
10
11pub use allocation::{
12 IcSnapshotAllocationExecutionError, IcSnapshotAllocationSettlementError, allocate_snapshot,
13};
14
15use crate::{
16 model::{
17 ic_mutation::IcMutationAcknowledgement,
18 ic_snapshot_upload::{
19 IcSnapshotUploadAttempt, IcSnapshotUploadAttemptError, IcSnapshotUploadRequest,
20 original_authority,
21 },
22 operation_plan::OperationPlanError,
23 },
24 ops::persistence::{
25 AttemptJournalError, AttemptJournalGuard, ExecutionProgressPersistenceError,
26 ExecutionStageGuard, ExecutionWorkflowPersistenceError,
27 },
28 policy::ic_snapshot_upload::{
29 IcSnapshotUploadAcknowledgementView, IcSnapshotUploadAssociationError,
30 validate_acknowledgement,
31 },
32 ports::{ic_mutation::IcMutationProviderError, ic_snapshot_upload::IcSnapshotUploadProvider},
33};
34use thiserror::Error;
35
36async fn upload_snapshot_with<E: std::error::Error + 'static, T>(
39 stage: &ExecutionStageGuard<'_>,
40 operation_sequence: u64,
41 payload: &IcSnapshotUploadRequest<'_>,
42 provider: &mut impl IcSnapshotUploadProvider,
43 admit: impl AsyncFnOnce(&IcSnapshotUploadAttempt<'_, '_>) -> Result<(), E>,
44 complete: impl AsyncFnOnce(
45 &mut AttemptJournalGuard<'_>,
46 &IcSnapshotUploadAttempt<'_, '_>,
47 &IcSnapshotUploadAcknowledgementView<'_, '_>,
48 ) -> T,
49) -> Result<(IcMutationAcknowledgement, T), IcSnapshotUploadExecutionError<E>> {
50 let plan = stage.plan();
51 let authority = plan.attempt_authority(operation_sequence)?;
52 let mut journal = AttemptJournalGuard::open(stage.layout()?, &authority)?;
53 original_authority(plan, operation_sequence, journal.record()?, payload)?;
54 journal.reserve_planned_mutation(&plan.digest())?;
55 let request =
56 IcSnapshotUploadAttempt::new(plan, operation_sequence, journal.record()?, payload)?;
57 admit(&request)
58 .await
59 .map_err(IcSnapshotUploadExecutionError::Admission)?;
60 stage.layout()?;
61 let acknowledgement = provider.submit_upload(&request, journal.record()?).await?;
62 let admit_reply = || {
63 let admitted = validate_acknowledgement(&request, journal.record()?, &acknowledgement)?;
64 stage.layout()?;
65 Ok::<_, IcSnapshotUploadReplyError>(admitted)
66 };
67 let admitted = match admit_reply() {
68 Ok(admitted) => admitted,
69 Err(source) => {
70 return Err(IcSnapshotUploadExecutionError::AfterReply {
71 source,
72 acknowledgement: Box::new(acknowledgement),
73 });
74 }
75 };
76 let result = complete(&mut journal, &request, &admitted).await;
77 Ok((acknowledgement, result))
78}
79
80#[derive(Debug, Error)]
82pub enum IcSnapshotUploadExecutionError<E: std::error::Error + 'static> {
83 #[error(transparent)]
85 Plan(#[from] OperationPlanError),
86 #[error(transparent)]
88 Stage(#[from] ExecutionWorkflowPersistenceError),
89 #[error(transparent)]
91 Journal(#[from] AttemptJournalError),
92 #[error(transparent)]
94 Progress(#[from] ExecutionProgressPersistenceError),
95 #[error(transparent)]
97 Request(#[from] IcSnapshotUploadAttemptError),
98 #[error("fresh snapshot upload admission failed: {0}")]
100 Admission(#[source] E),
101 #[error(transparent)]
103 Provider(#[from] IcMutationProviderError),
104 #[error("snapshot upload reply rejected: {source}")]
106 AfterReply {
107 source: IcSnapshotUploadReplyError,
109 acknowledgement: Box<IcMutationAcknowledgement>,
111 },
112}
113
114#[derive(Debug, Error)]
116pub enum IcSnapshotUploadReplyError {
117 #[error(transparent)]
119 Journal(#[from] AttemptJournalError),
120 #[error(transparent)]
122 Association(#[from] IcSnapshotUploadAssociationError),
123 #[error(transparent)]
125 Stage(#[from] ExecutionWorkflowPersistenceError),
126}
127
128#[cfg(all(test, unix))]
129mod tests;