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

ic_backup/workflow/ic_snapshot_metadata/
mod.rs
