ic_backup/workflow/ic_snapshot_upload/data/
mod.rs1use super::{IcSnapshotUploadExecutionError, upload_snapshot_with};
4use crate::{
5 model::{
6 attempt_journal::MutationReceiptRequest,
7 evidence_digest::EvidenceDigest,
8 execution_workflow::ExecutionStagePredecessorRecord,
9 ic_mutation::IcMutationAcknowledgement,
10 ic_snapshot_upload::{
11 IcSnapshotDataUploadPlan, IcSnapshotDataUploadPlanningError, IcSnapshotUploadAttempt,
12 },
13 },
14 ops::persistence::{
15 AttemptJournalError, DownloadJournalGuard, ExecutionProgressPersistenceError,
16 ExecutionStageCheckpointError, ExecutionStageGuard, ExecutionWorkflowPersistenceError,
17 IcSnapshotUploadArtifactError, read_execution_progress,
18 },
19 ports::ic_snapshot_upload::IcSnapshotUploadProvider,
20};
21use thiserror::Error;
22
23pub async fn upload_snapshot_data<E: std::error::Error + 'static>(
52 stage: &ExecutionStageGuard<'_>,
53 upload: &IcSnapshotDataUploadPlan<'_, '_, '_>,
54 source: &DownloadJournalGuard<'_>,
55 snapshot: &str,
56 provider: &mut impl IcSnapshotUploadProvider,
57 mut admit: impl AsyncFnMut(&IcSnapshotUploadAttempt<'_, '_>) -> Result<(), E>,
58 mut qualify: impl AsyncFnMut(
59 &IcSnapshotUploadAttempt<'_, '_>,
60 &IcMutationAcknowledgement,
61 ) -> Result<MutationReceiptRequest, E>,
62) -> Result<ExecutionStagePredecessorRecord, IcSnapshotDataUploadExecutionError<E>> {
63 upload.validate_binding(stage.binding())?;
64 let plan = upload
65 .plan()
66 .ok_or(IcSnapshotDataUploadPlanningError::NoDataWrites)?;
67 if stage.plan() != plan {
68 return Err(IcSnapshotDataUploadPlanningError::OriginalMismatch.into());
69 }
70 let progress = read_execution_progress(stage.layout()?, &plan.digest())?;
71 if !progress.is_unattempted() {
72 return Err(IcSnapshotDataUploadExecutionError::AlreadyAttempted);
73 }
74 let mut evidence = EvidenceDigest::new(b"ic-backup/ic-snapshot-data-upload/v1\0");
75 evidence.append(plan.digest().hash().as_bytes());
76 let session = source.upload_session(snapshot, upload)?;
77 for (index, kind) in upload.kinds().iter().enumerate() {
78 let sequence =
79 u64::try_from(index).map_err(|_| IcSnapshotDataUploadPlanningError::CountOverflow)?;
80 let payload = session.prepare(sequence, kind.clone())?;
81 let (acknowledgement, settlement) = upload_snapshot_with(
83 stage,
84 sequence,
85 &payload,
86 provider,
87 &mut admit,
88 async |journal, request, admitted| {
89 let receipt = qualify(request, admitted.acknowledgement())
90 .await
91 .map_err(IcSnapshotDataUploadReplyError::Qualification)?;
92 super::super::qualified_receipt::record_applied(
93 journal,
94 receipt,
95 request.mutation_attempt(),
96 payload.binding_digest().hash(),
97 IcSnapshotDataUploadReplyError::ReceiptRequired,
98 || {
99 stage.layout()?;
100 Ok(())
101 },
102 )?;
103 stage.layout()?;
104 Ok::<_, IcSnapshotDataUploadReplyError<E>>(admitted.reply().digest())
105 },
106 )
107 .await?;
108 let digest =
109 settlement.map_err(|source| IcSnapshotDataUploadExecutionError::AfterReply {
110 operation_sequence: sequence,
111 source,
112 acknowledgement: Box::new(acknowledgement),
113 })?;
114 evidence.append(digest.hash().as_bytes());
115 }
116 session.finish()?;
117 let progress = read_execution_progress(stage.layout()?, &plan.digest())?;
118 if !progress.all_applied() {
119 return Err(IcSnapshotDataUploadExecutionError::AlreadyAttempted);
120 }
121 Ok(stage.reconcile_checkpoint(evidence.finish())?)
122}
123
124#[derive(Debug, Error)]
126pub enum IcSnapshotDataUploadExecutionError<E: std::error::Error + 'static> {
127 #[error("snapshot data upload original stage was already attempted")]
129 AlreadyAttempted,
130 #[error(transparent)]
132 Planning(#[from] IcSnapshotDataUploadPlanningError),
133 #[error(transparent)]
135 Stage(#[from] ExecutionWorkflowPersistenceError),
136 #[error(transparent)]
138 Progress(#[from] ExecutionProgressPersistenceError),
139 #[error(transparent)]
141 Source(#[from] IcSnapshotUploadArtifactError),
142 #[error(transparent)]
144 Upload(#[from] IcSnapshotUploadExecutionError<E>),
145 #[error("snapshot data upload reply rejected for operation {operation_sequence}: {source}")]
147 AfterReply {
148 operation_sequence: u64,
150 source: IcSnapshotDataUploadReplyError<E>,
152 acknowledgement: Box<IcMutationAcknowledgement>,
154 },
155 #[error(transparent)]
157 Checkpoint(#[from] ExecutionStageCheckpointError),
158}
159
160#[derive(Debug, Error)]
162pub enum IcSnapshotDataUploadReplyError<E: std::error::Error + 'static> {
163 #[error("snapshot data upload qualification failed: {0}")]
165 Qualification(#[source] E),
166 #[error("snapshot data upload requires an explicit original Applied receipt")]
168 ReceiptRequired,
169 #[error(transparent)]
171 Stage(#[from] ExecutionWorkflowPersistenceError),
172 #[error(transparent)]
174 Journal(#[from] AttemptJournalError),
175}