laminar_core/cluster/control/leader_lease/
topology_restore.rs

1//! Read-only authorization for private target restore preparation under the held parent cut.
2
3use super::topology_admission::{
4    topology_assignment_error, topology_checkpoint_error, CONTROL_TIMEOUT,
5};
6use super::topology_migration_root::validate_manifest_budget;
7use super::{AssignmentSnapshotStore, LeaderLeaseStore};
8use crate::cluster::control::{
9    LocalProcessAuthorityIdentity, ProcessLeaseAuthority, TopologyAdmissionPhase, TopologyError,
10    TopologyOperationId, TopologyRestoreInput,
11};
12
13impl LeaderLeaseStore {
14    /// Read exact target restore input for a frozen, currently certified participant.
15    /// Call through the controller with its configured authorities. This grants only private
16    /// state preparation and never source start, actor installation, output or intake release.
17    ///
18    /// # Errors
19    /// Requires `CutPrepared`, a published root, the admitting leader, complete current process
20    /// certificates and unchanged assignment. Rejects damaged evidence or a 15 second deadline.
21    pub async fn topology_restore_input(
22        &self,
23        assignments: &AssignmentSnapshotStore,
24        processes: &ProcessLeaseAuthority,
25        operation_id: TopologyOperationId,
26        process: LocalProcessAuthorityIdentity,
27    ) -> Result<TopologyRestoreInput, TopologyError> {
28        tokio::time::timeout(CONTROL_TIMEOUT, async {
29            let current = self.load_record().await?.ok_or(TopologyError::Fenced)?;
30            let operation = current
31                .topology_operations
32                .iter()
33                .find(|entry| entry.operation_id == operation_id)
34                .ok_or_else(|| TopologyError::Conflict("unknown topology operation".into()))?;
35            if operation.phase != TopologyAdmissionPhase::CutPrepared {
36                return Err(TopologyError::Conflict(
37                    "target restore requires the held CutPrepared operation".into(),
38                ));
39            }
40            if !current.lease.matches_proof(&operation.admitted_by) {
41                return Err(TopologyError::Fenced);
42            }
43            self.audit_topology_operation(operation).await?;
44            let plan = self.load_topology_plan(&operation.plan).await?;
45            let descriptor = self
46                .require_topology_prepared(operation, &plan, processes)
47                .await?;
48            if operation.preparation.as_ref().is_none_or(|preparation| {
49                !preparation.certificates.iter().any(|certificate| {
50                    certificate.participant == process.participant
51                        && certificate.process_term == process.process_term
52                })
53            }) {
54                return Err(TopologyError::Fenced);
55            }
56            self.require_topology_process(processes, process).await?;
57            let assignment = assignments
58                .load()
59                .await
60                .map_err(topology_assignment_error)?
61                .ok_or(TopologyError::Fenced)?;
62            if assignment.draining
63                || assignment
64                    .assignment_fence()
65                    .map_err(|e| TopologyError::Invalid(e.to_string()))?
66                    != plan.assignment
67            {
68                return Err(TopologyError::Fenced);
69            }
70            self.reject_consumed_checkpoint_assignment(&current, &plan.assignment)
71                .await
72                .map_err(topology_checkpoint_error)?;
73            let binding = operation.migration_root.as_ref().ok_or_else(|| {
74                TopologyError::Conflict(
75                    "target restore requires the published migration root".into(),
76                )
77            })?;
78            let root = self.load_topology_root(&binding.root).await?;
79            let (outcome, checkpoint) = self
80                .cluster_outcome_with_committed_checkpoint(root.cut.checkpoint.epoch)
81                .await
82                .map_err(topology_checkpoint_error)?
83                .ok_or_else(|| TopologyError::Invalid("restore cut outcome is missing".into()))?;
84            let checkpoint = checkpoint
85                .ok_or_else(|| TopologyError::Invalid("restore cut is not committed".into()))?;
86            validate_parent_cut(&outcome, &checkpoint, &root, &plan, &descriptor)?;
87            let parent = self.load_catalog_manifest(&plan.parent_manifest).await?;
88            let target = self.load_catalog_manifest(&plan.target_manifest).await?;
89            // Renewals and target preparation receipts can advance while this read runs.
90            // Restore requirements, leader and assignment must stay exact. This writes no receipt.
91            self.require_topology_prepared(operation, &plan, processes)
92                .await?;
93            let after = self.load_record().await?.ok_or(TopologyError::Fenced)?;
94            if !after.lease.matches_proof(&operation.admitted_by)
95                || after
96                    .topology_operations
97                    .iter()
98                    .find(|entry| entry.operation_id == operation_id)
99                    .is_none_or(|entry| !operation.same_restore_binding(entry))
100                || assignments
101                    .load()
102                    .await
103                    .map_err(topology_assignment_error)?
104                    .as_ref()
105                    != Some(&assignment)
106            {
107                return Err(TopologyError::Fenced);
108            }
109            let owned_vnodes = assignment
110                .vnodes
111                .iter()
112                .filter_map(|(vnode, owner)| {
113                    (owner.0 == process.participant.node_id).then_some(*vnode)
114                })
115                .collect();
116            Ok(TopologyRestoreInput {
117                operation: operation.clone(),
118                restore_assignment: plan.assignment.clone(),
119                restore_processes: Vec::new(),
120                committed_leader: None,
121                plan,
122                parent,
123                target,
124                descriptor,
125                root,
126                outcome,
127                checkpoint,
128                owned_vnodes,
129                process,
130            })
131        })
132        .await
133        .map_err(|_| TopologyError::ReadTimedOut)?
134    }
135}
136
137pub(super) fn validate_parent_cut(
138    outcome: &crate::checkpoint_decision::CheckpointOutcome,
139    checkpoint: &crate::checkpoint::CommittedCheckpointIndex,
140    root: &crate::cluster::control::TopologyMigrationRoot,
141    plan: &crate::cluster::control::TopologyAdmissionPlan,
142    descriptor: &crate::cluster::control::ClusterTopologyValidation,
143) -> Result<(), TopologyError> {
144    if !outcome.is_commit()
145        || outcome.committed_checkpoint.as_ref() != Some(&root.cut.checkpoint)
146        || checkpoint.pipeline_identity != descriptor.parent_pipeline
147        || checkpoint.deployment_id != descriptor.deployment_id
148        || checkpoint.assignment_fence.as_ref() != Some(&plan.assignment)
149    {
150        return Err(TopologyError::Invalid(
151            "topology restore differs from the exact historical parent cut".into(),
152        ));
153    }
154    validate_manifest_budget(checkpoint)
155}
laminar_core/cluster/control/leader_lease/topology_restore.rs

laminar_core/cluster/control/leader_lease/
topology_restore.rs
