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

ic_backup/workflow/ic_snapshot_upload/allocation/
mod.rs
