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::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 metadata_digest = super::qualified_receipt::AppliedReceiptRecorder {
118        stage,
119        journal: &mut journal,
120        expected_attempt: request.mutation_attempt(),
121        expected_request: payload.digest().hash(),
122    }
123    .record_applied(
124        &request,
125        response,
126        async |request, reply| qualify(request, reply).await,
127        IcSnapshotMetadataSettlementError::Qualification,
128        || IcSnapshotMetadataSettlementError::ReceiptRequired,
129        || match admitted.reply() {
130            IcSnapshotTransferReadReply::Metadata(metadata) => Ok(metadata.digest()),
131            IcSnapshotTransferReadReply::Data(_) => {
132                Err(IcSnapshotMetadataSettlementError::ReceiptRequired)
133            }
134        },
135    )
136    .await?;
137    drop(journal);
138    Ok(stage.checkpoint(metadata_digest)?)
139}
140
141/// Metadata coordination failures retain exact original spending and returned evidence.
142#[derive(Debug, Error)]
143pub enum IcSnapshotMetadataExecutionError<E: std::error::Error + 'static> {
144    /// The original plan is not exactly this single metadata operation.
145    #[error("snapshot metadata requires an exact singleton original stage")]
146    OriginalMismatch,
147    /// Canonical single-read failure, including any bounded returned response.
148    #[error(transparent)]
149    Read(#[from] IcSnapshotTransferReadExecutionError<E>),
150    /// Receipt or checkpoint failed after a response; all originals remain retained.
151    #[error("snapshot metadata reply settlement failed: {source}")]
152    AfterReply {
153        /// Exact original rejection; no retry or refund follows.
154        source: IcSnapshotMetadataSettlementError<E>,
155        /// Original bounded reply and passive claims.
156        response: Box<IcSnapshotTransferReadResponse>,
157    },
158}
159
160/// Rejection after one returned metadata reply; no second accounting owner.
161#[derive(Debug, Error)]
162pub enum IcSnapshotMetadataSettlementError<E: std::error::Error + 'static> {
163    /// Actual authentication, attribution or durable original-byte retention failed.
164    #[error("snapshot metadata qualification failed: {0}")]
165    Qualification(#[source] E),
166    /// Require an independently qualified exact original Applied receipt.
167    #[error("snapshot metadata requires an explicit original Applied receipt")]
168    ReceiptRequired,
169    /// Original stage or retained ancestor admission failed.
170    #[error(transparent)]
171    Stage(#[from] ExecutionWorkflowPersistenceError),
172    /// Original selected journal or receipt admission failed.
173    #[error(transparent)]
174    Journal(#[from] AttemptJournalError),
175    /// Exact original request/current reservation admission failed.
176    #[error(transparent)]
177    Request(#[from] IcSnapshotTransferReadError),
178    /// Existing bounded passive response admission failed.
179    #[error(transparent)]
180    Association(#[from] IcSnapshotTransferReadAssociationError),
181    /// Canonical all-Applied checkpoint publication/re-admission failed.
182    #[error(transparent)]
183    Checkpoint(#[from] ExecutionStageCheckpointError),
184}
185
186#[cfg(all(test, unix))]
187mod tests;
ic_backup/workflow/ic_snapshot_metadata/mod.rs

ic_backup/workflow/ic_snapshot_metadata/
mod.rs
