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        journal: &mut journal,
111        expected_attempt: request.mutation_attempt(),
112        expected_request: payload.binding_digest().hash(),
113    }
114    .record_applied(
115        &request,
116        acknowledgement,
117        async |request, reply| qualify(request, reply).await,
118        IcSnapshotAllocationSettlementError::Qualification,
119        || IcSnapshotAllocationSettlementError::ReceiptRequired,
120        || {
121            stage
122                .layout()
123                .map(|_| ())
124                .map_err(IcSnapshotAllocationSettlementError::Stage)
125        },
126    )
127    .await?;
128    drop(journal);
129    Ok(stage.checkpoint(admitted.reply().digest())?)
130}
131
132/// Allocation coordination failures retain original spending and bounded returned bytes.
133#[derive(Debug, Error)]
134pub enum IcSnapshotAllocationExecutionError<E: std::error::Error + 'static> {
135    /// Original plan is not exactly this singleton metadata allocation.
136    #[error("snapshot allocation requires an exact singleton original metadata stage")]
137    OriginalMismatch,
138    /// Canonical single-upload failure, including bounded returned acknowledgements.
139    #[error(transparent)]
140    Upload(#[from] IcSnapshotUploadExecutionError<E>),
141    /// Receipt or checkpoint failed after reply; all originals remain retained.
142    #[error("snapshot allocation reply settlement failed: {source}")]
143    AfterReply {
144        /// Exact settlement rejection; no retry or refund follows.
145        source: IcSnapshotAllocationSettlementError<E>,
146        /// Original bounded reply and passive claims.
147        acknowledgement: Box<IcMutationAcknowledgement>,
148    },
149}
150
151/// Post-reply allocation rejection; no second accounting or outcome owner.
152#[derive(Debug, Error)]
153pub enum IcSnapshotAllocationSettlementError<E: std::error::Error + 'static> {
154    /// Independent authentication, attribution or durable original-byte retention failed.
155    #[error("snapshot allocation qualification failed: {0}")]
156    Qualification(#[source] E),
157    /// Require an independently qualified exact original Applied receipt.
158    #[error("snapshot allocation requires an explicit original Applied receipt")]
159    ReceiptRequired,
160    /// Original stage or ancestor admission failed.
161    #[error(transparent)]
162    Stage(#[from] ExecutionWorkflowPersistenceError),
163    /// Selected journal or receipt admission failed.
164    #[error(transparent)]
165    Journal(#[from] AttemptJournalError),
166    /// Original upload/current reservation admission failed.
167    #[error(transparent)]
168    Request(#[from] IcSnapshotUploadAttemptError),
169    /// Canonical bounded acknowledgement association failed.
170    #[error(transparent)]
171    Association(#[from] IcSnapshotUploadAssociationError),
172    /// Canonical all-Applied checkpoint publication/re-admission failed.
173    #[error(transparent)]
174    Checkpoint(#[from] ExecutionStageCheckpointError),
175}
176
177#[cfg(all(test, unix))]
178mod tests;
ic_backup/workflow/ic_snapshot_upload/allocation/mod.rs

ic_backup/workflow/ic_snapshot_upload/allocation/
mod.rs
