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

ic_backup/workflow/ic_snapshot_metadata/
mod.rs
