ic_backup/workflow/ic_snapshot_upload/mod.rs
1//! One fresh original source-bound metadata/data upload; no automatic receipt or retry.
2
3mod 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::{IcSnapshotUploadAssociationError, validate_acknowledgement},
29 ports::{ic_mutation::IcMutationProviderError, ic_snapshot_upload::IcSnapshotUploadProvider},
30};
31use thiserror::Error;
32
33/// Durably reserve one exact original upload, freshly admit it and submit once.
34///
35/// Prepare the retained stage and every original journal before entry; hold no attempt
36/// guards. Guarded source preparation and exact metadata/tree/payload retention precede
37/// the upload plan. Data bytes and their independently attributed destination must be
38/// bound before that data stage is published, within its original workflow allocation.
39/// This checks the canonical full source/upload context and binding before spending,
40/// then delegates reservation/prerequisites to the existing complete-plan owner.
41/// Missing, pending, Applied or exhausted originals never reach provider invocation.
42///
43/// The mandatory fallible `admit` callback qualifies authentic complete source bytes,
44/// actual fresh controllers and application/restore requirements, exact metadata/data
45/// destination attribution, stable source/payload custody and exclusive never-dispatched
46/// command custody. It runs under the selected journal lock; remote preflight requires
47/// its own prior accounting. Local checks grant no byte fence or allocation authority.
48/// The existing provider sends only the exact accounted replicated update, with no
49/// batching, retries, hidden observations or implicit funding. No default is installed.
50/// Await admission and submission under the selected original guard. Cancellation
51/// releases exclusion without changing spending or permitting reentry. The provider
52/// receives the currently guarded original journal; the core supplies no executor.
53///
54/// Re-admit original stage/ancestor evidence before and after dispatch and check the
55/// bounded acknowledgement through the canonical upload policy/decoder. Success
56/// remains pending; integrations independently authenticate attribution and retain
57/// exact replies before explicitly recording receipts with the existing journal owner.
58/// A returned ID/empty reply alone grants no allocation outcome, learned data authority,
59/// complete transfer, load/start admission, terminal or fence/source-reference release.
60/// # Errors
61/// Rejects changed originals/payload/context, spending, fresh admission, provider or
62/// reply failures. Post-reservation failures retain consumption; post-reply rejections
63/// retain the bounded acknowledgement. No failure grants redispatch/refund/cleanup.
64pub async fn upload_snapshot<E: std::error::Error + 'static>(
65 stage: &ExecutionStageGuard<'_>,
66 operation_sequence: u64,
67 payload: &IcSnapshotUploadRequest<'_>,
68 provider: &mut impl IcSnapshotUploadProvider,
69 admit: impl AsyncFnOnce(&IcSnapshotUploadAttempt<'_, '_>) -> Result<(), E>,
70) -> Result<IcMutationAcknowledgement, IcSnapshotUploadExecutionError<E>> {
71 let plan = stage.plan();
72 let authority = plan.attempt_authority(operation_sequence)?;
73 let mut journal = AttemptJournalGuard::open(stage.layout()?, &authority)?;
74 original_authority(plan, operation_sequence, journal.record()?, payload)?;
75 journal.reserve_planned_mutation(&plan.digest())?;
76 let request =
77 IcSnapshotUploadAttempt::new(plan, operation_sequence, journal.record()?, payload)?;
78 let acknowledgement = super::reserved_effect::submit_once(
79 &journal,
80 &request,
81 async |request| {
82 admit(request)
83 .await
84 .map_err(IcSnapshotUploadExecutionError::Admission)
85 },
86 || {
87 stage
88 .layout()
89 .map(|_| ())
90 .map_err(IcSnapshotUploadExecutionError::Stage)
91 },
92 async |request, record| {
93 provider
94 .submit_upload(request, record)
95 .await
96 .map_err(IcSnapshotUploadExecutionError::Provider)
97 },
98 IcSnapshotUploadExecutionError::Journal,
99 )
100 .await?;
101 if let Err(error) = super::reserved_effect::check_returned(
102 stage,
103 &journal,
104 &request,
105 &acknowledgement,
106 |request, record, reply| validate_acknowledgement(request, record, reply).map(|_| ()),
107 ) {
108 return Err(match error {
109 super::reserved_effect::ReturnedError::Journal(source) => {
110 IcSnapshotUploadExecutionError::AfterReplyJournal {
111 source,
112 acknowledgement: Box::new(acknowledgement),
113 }
114 }
115 super::reserved_effect::ReturnedError::Association(source) => {
116 IcSnapshotUploadExecutionError::Association {
117 source,
118 acknowledgement: Box::new(acknowledgement),
119 }
120 }
121 super::reserved_effect::ReturnedError::Stage(source) => {
122 IcSnapshotUploadExecutionError::AfterReplyStage {
123 source,
124 acknowledgement: Box::new(acknowledgement),
125 }
126 }
127 });
128 }
129 Ok(acknowledgement)
130}
131
132/// Upload-step failures retain original spending and any bounded returned acknowledgement.
133#[derive(Debug, Error)]
134pub enum IcSnapshotUploadExecutionError<E: std::error::Error + 'static> {
135 /// Original operation authority cannot be derived.
136 #[error(transparent)]
137 Plan(#[from] OperationPlanError),
138 /// Workflow, stage or original ancestor evidence failed admission.
139 #[error(transparent)]
140 Stage(#[from] ExecutionWorkflowPersistenceError),
141 /// Original selected journal admission failed.
142 #[error(transparent)]
143 Journal(#[from] AttemptJournalError),
144 /// Complete original progress, prerequisites or durable reservation failed.
145 #[error(transparent)]
146 Progress(#[from] ExecutionProgressPersistenceError),
147 /// Canonical original mutation binding or current reservation differs.
148 #[error(transparent)]
149 Request(#[from] IcSnapshotUploadAttemptError),
150 /// Fresh integration-owned admission rejected after reservation.
151 #[error("fresh snapshot upload admission failed: {0}")]
152 Admission(#[source] E),
153 /// The single provider call failed; spending remains pending.
154 #[error(transparent)]
155 Provider(#[from] IcMutationProviderError),
156 /// Passive bounded association rejected a returned acknowledgement.
157 #[error("snapshot upload acknowledgement association failed: {source}")]
158 Association {
159 /// Existing canonical mutation association rejection.
160 source: IcSnapshotUploadAssociationError,
161 /// Exact returned acknowledgement, without authenticated outcome.
162 acknowledgement: Box<IcMutationAcknowledgement>,
163 },
164 /// Selected journal re-admission failed after a reply.
165 #[error("snapshot upload journal changed after reply: {source}")]
166 AfterReplyJournal {
167 /// Original journal rejection.
168 source: AttemptJournalError,
169 /// Exact bounded returned acknowledgement.
170 acknowledgement: Box<IcMutationAcknowledgement>,
171 },
172 /// Stage or ancestor re-admission failed after a reply.
173 #[error("snapshot upload stage changed after reply: {source}")]
174 AfterReplyStage {
175 /// Original stage or ancestor rejection.
176 source: ExecutionWorkflowPersistenceError,
177 /// Exact bounded returned acknowledgement.
178 acknowledgement: Box<IcMutationAcknowledgement>,
179 },
180}
181
182#[cfg(all(test, unix))]
183mod tests;