ic_backup/workflow/ic_snapshot_metadata/
mod.rs

1//! One original metadata stage, with explicit qualified receipt and learned checkpoint.
2
3use crate::{
4    model::{
5        attempt_journal::{MutationOutcomeRecord, MutationReceiptRequest},
6        execution_workflow::ExecutionStagePredecessorRecord,
7        ic_snapshot_metadata::IcSnapshotMetadataRequest,
8        ic_snapshot_transfer_read::{
9            IcSnapshotTransferReadError, IcSnapshotTransferReadPayload,
10            IcSnapshotTransferReadRequest, IcSnapshotTransferReadResponse,
11        },
12    },
13    ops::persistence::{
14        AttemptJournalError, AttemptJournalGuard, ExecutionStageCheckpointError,
15        ExecutionStageGuard, ExecutionWorkflowPersistenceError,
16    },
17    policy::ic_snapshot_transfer_read::{
18        IcSnapshotTransferReadAssociationError, IcSnapshotTransferReadReply, validate_response,
19    },
20    ports::ic_snapshot_transfer_read::IcSnapshotTransferReadProvider,
21    workflow::ic_snapshot_transfer_read::{IcSnapshotTransferReadExecutionError, read_snapshot},
22};
23use thiserror::Error;
24
25/// Read one new exact singleton metadata stage, record its qualified receipt and checkpoint.
26///
27/// Prepare the original stage and complete journals first; hold no attempt guards on
28/// entry. Reuse the canonical single-read owner for durable spending, mandatory fresh
29/// admission and exactly one provider call. A pending or previously Applied read is
30/// never dispatched again, including after interruption before checkpoint publication.
31///
32/// `admit` owns fresh read permission, raw-ID/application and never-dispatched custody.
33/// `qualify` runs under the original journal lock and must independently authenticate
34/// exact response attribution AND durably retain original request/reply bytes before
35/// returning the explicit Applied receipt. There is no default qualification or byte
36/// store. Opaque provider evidence and valid wire shape cannot substitute for it.
37/// Account any remote qualification separately before its calls.
38/// Admission, submission and independent qualification are async. Qualification
39/// holds the selected original journal and must retain the reply durably before
40/// awaiting cancellable work. Cancellation keeps pending/Applied spending and denies
41/// reentry; no executor or `Send` bound is imposed by the core.
42///
43/// Record the receipt through the sole spending owner, release its lock, then publish
44/// the canonical all-Applied checkpoint with the exact request/raw-metadata digest as
45/// learned evidence. The returned response and predecessor can feed the existing
46/// metadata decoder and `IcSnapshotDownloadPlan::bind`; preparing the data stage and
47/// writer remains explicit. Failure retains spending, receipts, checkpoints and bytes;
48/// post-read errors retain the bounded response. Reopen/checkpoint recovery never
49/// repeats a read. No data calls, artifact, default Agent bridge, complete backup,
50/// terminal proof or fence/reference release follows.
51/// # Errors
52/// Rejects non-singleton or changed originals, spending, fresh admission, provider,
53/// association, qualification, receipt and checkpoint failures without cleanup/refund.
54pub async fn read_snapshot_metadata<E: std::error::Error + 'static>(
55    stage: &ExecutionStageGuard<'_>,
56    operation_sequence: u64,
57    payload: &IcSnapshotMetadataRequest,
58    provider: &mut impl IcSnapshotTransferReadProvider,
59    admit: impl AsyncFnOnce(&IcSnapshotTransferReadRequest<'_, '_>) -> Result<(), E>,
60    qualify: impl AsyncFnOnce(
61        &IcSnapshotTransferReadRequest<'_, '_>,
62        &IcSnapshotTransferReadResponse,
63    ) -> Result<MutationReceiptRequest, E>,
64) -> Result<
65    (
66        IcSnapshotTransferReadResponse,
67        ExecutionStagePredecessorRecord,
68    ),
69    IcSnapshotMetadataExecutionError<E>,
70> {
71    let operations = stage.plan().operations();
72    if operations.len() != 1
73        || operations[0].operation_sequence() != operation_sequence
74        || operations[0].request() != payload.digest().hash()
75    {
76        return Err(IcSnapshotMetadataExecutionError::OriginalMismatch);
77    }
78    let response = read_snapshot(
79        stage,
80        operation_sequence,
81        IcSnapshotTransferReadPayload::Metadata(payload),
82        provider,
83        admit,
84    )
85    .await?;
86    match settle_metadata(stage, operation_sequence, payload, &response, qualify).await {
87        Ok(predecessor) => Ok((response, predecessor)),
88        Err(source) => Err(IcSnapshotMetadataExecutionError::AfterReply {
89            source,
90            response: Box::new(response),
91        }),
92    }
93}
94
95async fn settle_metadata<E: std::error::Error + 'static>(
96    stage: &ExecutionStageGuard<'_>,
97    sequence: u64,
98    payload: &IcSnapshotMetadataRequest,
99    response: &IcSnapshotTransferReadResponse,
100    qualify: impl AsyncFnOnce(
101        &IcSnapshotTransferReadRequest<'_, '_>,
102        &IcSnapshotTransferReadResponse,
103    ) -> Result<MutationReceiptRequest, E>,
104) -> Result<ExecutionStagePredecessorRecord, IcSnapshotMetadataSettlementError<E>> {
105    let plan = stage.plan();
106    let authority = plan
107        .attempt_authority(sequence)
108        .map_err(IcSnapshotTransferReadError::from)?;
109    let mut journal = AttemptJournalGuard::open(stage.layout()?, &authority)?;
110    let request = IcSnapshotTransferReadRequest::new(
111        plan,
112        sequence,
113        journal.record()?,
114        IcSnapshotTransferReadPayload::Metadata(payload),
115    )?;
116    let admitted = validate_response(&request, journal.record()?, response)?;
117    let receipt = qualify(&request, response)
118        .await
119        .map_err(IcSnapshotMetadataSettlementError::Qualification)?;
120    if receipt.outcome != MutationOutcomeRecord::Applied
121        || receipt.attempt != request.mutation_attempt()
122        || receipt.request != payload.digest().hash()
123    {
124        return Err(IcSnapshotMetadataSettlementError::ReceiptRequired);
125    }
126    let IcSnapshotTransferReadReply::Metadata(metadata) = admitted.reply() else {
127        return Err(IcSnapshotMetadataSettlementError::ReceiptRequired);
128    };
129    stage.layout()?;
130    journal.record_mutation(receipt)?;
131    drop(journal);
132    Ok(stage.checkpoint(metadata.digest())?)
133}
134
135/// Metadata coordination failures retain exact original spending and returned evidence.
136#[derive(Debug, Error)]
137pub enum IcSnapshotMetadataExecutionError<E: std::error::Error + 'static> {
138    /// The original plan is not exactly this single metadata operation.
139    #[error("snapshot metadata requires an exact singleton original stage")]
140    OriginalMismatch,
141    /// Canonical single-read failure, including any bounded returned response.
142    #[error(transparent)]
143    Read(#[from] IcSnapshotTransferReadExecutionError<E>),
144    /// Receipt or checkpoint failed after a response; all originals remain retained.
145    #[error("snapshot metadata reply settlement failed: {source}")]
146    AfterReply {
147        /// Exact original rejection; no retry or refund follows.
148        source: IcSnapshotMetadataSettlementError<E>,
149        /// Original bounded reply and passive claims.
150        response: Box<IcSnapshotTransferReadResponse>,
151    },
152}
153
154/// Rejection after one returned metadata reply; no second accounting owner.
155#[derive(Debug, Error)]
156pub enum IcSnapshotMetadataSettlementError<E: std::error::Error + 'static> {
157    /// Actual authentication, attribution or durable original-byte retention failed.
158    #[error("snapshot metadata qualification failed: {0}")]
159    Qualification(#[source] E),
160    /// Require an independently qualified exact original Applied receipt.
161    #[error("snapshot metadata requires an explicit original Applied receipt")]
162    ReceiptRequired,
163    /// Original stage or retained ancestor admission failed.
164    #[error(transparent)]
165    Stage(#[from] ExecutionWorkflowPersistenceError),
166    /// Original selected journal or receipt admission failed.
167    #[error(transparent)]
168    Journal(#[from] AttemptJournalError),
169    /// Exact original request/current reservation admission failed.
170    #[error(transparent)]
171    Request(#[from] IcSnapshotTransferReadError),
172    /// Existing bounded passive response admission failed.
173    #[error(transparent)]
174    Association(#[from] IcSnapshotTransferReadAssociationError),
175    /// Canonical all-Applied checkpoint publication/re-admission failed.
176    #[error(transparent)]
177    Checkpoint(#[from] ExecutionStageCheckpointError),
178}
179
180#[cfg(all(test, unix))]
181mod tests;
ic_backup/workflow/ic_snapshot_metadata/mod.rs

ic_backup/workflow/ic_snapshot_metadata/
mod.rs
