ic_backup/workflow/ic_snapshot_upload/allocation/
mod.rs

1//! Original metadata allocation with independently qualified receipt and learned checkpoint.
2
3use super::{IcSnapshotUploadExecutionError, upload_snapshot};
4use crate::{
5    model::{
6        attempt_journal::MutationReceiptRequest,
7        execution_workflow::ExecutionStagePredecessorRecord,
8        ic_mutation::IcMutationAcknowledgement,
9        ic_snapshot_upload::{
10            IcSnapshotUploadAttempt, IcSnapshotUploadAttemptError, IcSnapshotUploadKind,
11            IcSnapshotUploadRequest,
12        },
13    },
14    ops::persistence::{
15        AttemptJournalError, AttemptJournalGuard, ExecutionStageCheckpointError,
16        ExecutionStageGuard, ExecutionWorkflowPersistenceError,
17    },
18    policy::ic_snapshot_upload::{IcSnapshotUploadAssociationError, validate_acknowledgement},
19    ports::ic_snapshot_upload::IcSnapshotUploadProvider,
20};
21use thiserror::Error;
22
23/// Allocate one new snapshot through an exact singleton original metadata stage.
24///
25/// Prepare the retained workflow/stage and complete original journals first; hold no
26/// attempt guards. Reuse [`upload_snapshot`] for exact source/payload binding, durable
27/// spending, mandatory fresh `admit` and one provider invocation. A pending or Applied
28/// allocation never dispatches again, including after failed checkpoint publication.
29///
30/// The mandatory `qualify` callback runs under the original journal lock. It must
31/// independently authenticate exclusive allocation attribution and durably retain the
32/// original request/reply bytes before returning its exact Applied receipt. Neither a
33/// decoded destination ID nor provider evidence supplies that qualification. Account
34/// any remote qualification independently before its calls; no default is installed.
35/// Admission, submission and qualification are async under their selected journal
36/// guards. Retain exact replies before cancellable qualification work. Cancellation
37/// keeps pending/Applied spending and denies allocation reentry; no runtime or `Send`
38/// bound is imposed by the core.
39///
40/// Record only that exact qualified receipt through the sole attempt owner, release
41/// the journal lock, then checkpoint the canonical allocation request/raw-reply digest.
42/// Return the bounded acknowledgement and original predecessor for explicit data-stage
43/// binding within the original workflow ceiling. Original source/metadata custody,
44/// destination attribution and data planning remain integration-owned. No data call,
45/// replacement, complete transfer, load/start, terminal or fence/reference release
46/// follows. Recovery reopens retained evidence and never repeats this allocation.
47/// # Errors
48/// Rejects non-metadata/non-singleton originals, spending/admission/provider/reply,
49/// qualification, receipt and checkpoint failures. Post-reply failures retain bounded
50/// acknowledgement bytes; spending, recorded outcomes and occupied evidence remain.
51pub async fn allocate_snapshot<E: std::error::Error + 'static>(
52    stage: &ExecutionStageGuard<'_>,
53    operation_sequence: u64,
54    payload: &IcSnapshotUploadRequest<'_>,
55    provider: &mut impl IcSnapshotUploadProvider,
56    admit: impl AsyncFnOnce(&IcSnapshotUploadAttempt<'_, '_>) -> Result<(), E>,
57    qualify: impl AsyncFnOnce(
58        &IcSnapshotUploadAttempt<'_, '_>,
59        &IcMutationAcknowledgement,
60    ) -> Result<MutationReceiptRequest, E>,
61) -> Result<
62    (IcMutationAcknowledgement, ExecutionStagePredecessorRecord),
63    IcSnapshotAllocationExecutionError<E>,
64> {
65    let operations = stage.plan().operations();
66    if !matches!(payload.kind(), IcSnapshotUploadKind::Metadata)
67        || operations.len() != 1
68        || operations[0].operation_sequence() != operation_sequence
69        || operations[0].request() != payload.binding_digest().hash()
70    {
71        return Err(IcSnapshotAllocationExecutionError::OriginalMismatch);
72    }
73    let acknowledgement =
74        upload_snapshot(stage, operation_sequence, payload, provider, admit).await?;
75    match settle_allocation(
76        stage,
77        operation_sequence,
78        payload,
79        &acknowledgement,
80        qualify,
81    )
82    .await
83    {
84        Ok(predecessor) => Ok((acknowledgement, predecessor)),
85        Err(source) => Err(IcSnapshotAllocationExecutionError::AfterReply {
86            source,
87            acknowledgement: Box::new(acknowledgement),
88        }),
89    }
90}
91
92async fn settle_allocation<E: std::error::Error + 'static>(
93    stage: &ExecutionStageGuard<'_>,
94    sequence: u64,
95    payload: &IcSnapshotUploadRequest<'_>,
96    acknowledgement: &IcMutationAcknowledgement,
97    qualify: impl AsyncFnOnce(
98        &IcSnapshotUploadAttempt<'_, '_>,
99        &IcMutationAcknowledgement,
100    ) -> Result<MutationReceiptRequest, E>,
101) -> Result<ExecutionStagePredecessorRecord, IcSnapshotAllocationSettlementError<E>> {
102    let plan = stage.plan();
103    let authority = plan
104        .attempt_authority(sequence)
105        .map_err(IcSnapshotUploadAttemptError::from)?;
106    let mut journal = AttemptJournalGuard::open(stage.layout()?, &authority)?;
107    let request = IcSnapshotUploadAttempt::new(plan, sequence, journal.record()?, payload)?;
108    let admitted = validate_acknowledgement(&request, journal.record()?, acknowledgement)?;
109    super::super::qualified_receipt::AppliedReceiptRecorder {
110        stage,
111        journal: &mut journal,
112        expected_attempt: request.mutation_attempt(),
113        expected_request: payload.binding_digest().hash(),
114    }
115    .record_applied(
116        &request,
117        acknowledgement,
118        async |request, reply| qualify(request, reply).await,
119        IcSnapshotAllocationSettlementError::Qualification,
120        || IcSnapshotAllocationSettlementError::ReceiptRequired,
121        || Ok(()),
122    )
123    .await?;
124    drop(journal);
125    Ok(stage.checkpoint(admitted.reply().digest())?)
126}
127
128/// Allocation coordination failures retain original spending and bounded returned bytes.
129#[derive(Debug, Error)]
130pub enum IcSnapshotAllocationExecutionError<E: std::error::Error + 'static> {
131    /// Original plan is not exactly this singleton metadata allocation.
132    #[error("snapshot allocation requires an exact singleton original metadata stage")]
133    OriginalMismatch,
134    /// Canonical single-upload failure, including bounded returned acknowledgements.
135    #[error(transparent)]
136    Upload(#[from] IcSnapshotUploadExecutionError<E>),
137    /// Receipt or checkpoint failed after reply; all originals remain retained.
138    #[error("snapshot allocation reply settlement failed: {source}")]
139    AfterReply {
140        /// Exact settlement rejection; no retry or refund follows.
141        source: IcSnapshotAllocationSettlementError<E>,
142        /// Original bounded reply and passive claims.
143        acknowledgement: Box<IcMutationAcknowledgement>,
144    },
145}
146
147/// Post-reply allocation rejection; no second accounting or outcome owner.
148#[derive(Debug, Error)]
149pub enum IcSnapshotAllocationSettlementError<E: std::error::Error + 'static> {
150    /// Independent authentication, attribution or durable original-byte retention failed.
151    #[error("snapshot allocation qualification failed: {0}")]
152    Qualification(#[source] E),
153    /// Require an independently qualified exact original Applied receipt.
154    #[error("snapshot allocation requires an explicit original Applied receipt")]
155    ReceiptRequired,
156    /// Original stage or ancestor admission failed.
157    #[error(transparent)]
158    Stage(#[from] ExecutionWorkflowPersistenceError),
159    /// Selected journal or receipt admission failed.
160    #[error(transparent)]
161    Journal(#[from] AttemptJournalError),
162    /// Original upload/current reservation admission failed.
163    #[error(transparent)]
164    Request(#[from] IcSnapshotUploadAttemptError),
165    /// Canonical bounded acknowledgement association failed.
166    #[error(transparent)]
167    Association(#[from] IcSnapshotUploadAssociationError),
168    /// Canonical all-Applied checkpoint publication/re-admission failed.
169    #[error(transparent)]
170    Checkpoint(#[from] ExecutionStageCheckpointError),
171}
172
173#[cfg(all(test, unix))]
174mod tests;
ic_backup/workflow/ic_snapshot_upload/allocation/mod.rs

ic_backup/workflow/ic_snapshot_upload/allocation/
mod.rs
