laminar_core/cluster/control/leader_lease/
topology_recovery.rs

1//! Exact target checkpoint selection; damaged newer progress never falls back to a parent root.
2
3use super::topology_admission::{topology_checkpoint_error, CONTROL_TIMEOUT};
4use super::topology_migration_root::validate_manifest_budget;
5use super::{AssignmentSnapshotStore, LeaderLeaseStore};
6use crate::checkpoint::CheckpointScope;
7use crate::cluster::control::{
8    LocalProcessAuthorityIdentity, ProcessLeaseAuthority, TopologyAdmissionPhase, TopologyError,
9    TopologyOperationId, TopologyRecoveryInput,
10};
11
12impl LeaderLeaseStore {
13    /// Select the current target's greatest committed checkpoint, or its immutable migration root
14    /// when no target checkpoint has committed. Replacement boots must retain the owner map and
15    /// stable roster. This permits private reads only, never actors, sink epochs or intake.
16    ///
17    /// # Errors
18    /// Rejects stale process/assignment/leader, an obsolete target, damaged retained evidence or
19    /// any newer Commit that cannot be proved to belong to the target. No older-cut fallback is
20    /// permitted. The complete selection and authority recheck share a 15 second budget.
21    pub async fn committed_topology_recovery_input(
22        &self,
23        assignments: &AssignmentSnapshotStore,
24        processes: &ProcessLeaseAuthority,
25        operation_id: TopologyOperationId,
26        process: LocalProcessAuthorityIdentity,
27    ) -> Result<TopologyRecoveryInput, TopologyError> {
28        tokio::time::timeout(CONTROL_TIMEOUT, async {
29            let before = self.load_record().await?.ok_or(TopologyError::Fenced)?;
30            let migration = self
31                .committed_topology_restore_input(assignments, processes, operation_id, process)
32                .await?;
33            let operation = migration.operation();
34            let commit = operation.commit.as_ref().ok_or(TopologyError::Fenced)?;
35            if before.committed_topology_operation() != Some(operation)
36                || Some(before.lease.proof()) != migration.current_leader()
37            {
38                return Err(TopologyError::Fenced);
39            }
40            let head = before.commit_head.ok_or_else(|| {
41                TopologyError::Invalid("committed topology has no checkpoint Commit head".into())
42            })?;
43            let root = &migration.root().cut;
44            let target_checkpoint = if head.sequence == root.authority_sequence {
45                if head.epoch != root.checkpoint.epoch
46                    || head.checkpoint_id != root.checkpoint.checkpoint_id
47                {
48                    return Err(TopologyError::Invalid(
49                        "migration root differs from its current checkpoint Commit head".into(),
50                    ));
51                }
52                None
53            } else {
54                if head.sequence <= commit.authority_sequence
55                    || head.epoch <= root.checkpoint.epoch
56                    || operation.phase != TopologyAdmissionPhase::Active
57                {
58                    return Err(TopologyError::Invalid(
59                        "newer checkpoint Commit is not authorized target progress".into(),
60                    ));
61                }
62                let (outcome, checkpoint) = self
63                    .cluster_outcome_with_committed_checkpoint(head.epoch)
64                    .await
65                    .map_err(topology_checkpoint_error)?
66                    .ok_or_else(|| {
67                        TopologyError::Invalid(
68                            "selected target checkpoint outcome is missing".into(),
69                        )
70                    })?;
71                let checkpoint = checkpoint.ok_or_else(|| {
72                    TopologyError::Invalid("selected target checkpoint is not committed".into())
73                })?;
74                let assignment = checkpoint.assignment_fence.as_ref().ok_or_else(|| {
75                    TopologyError::Invalid("target checkpoint has no assignment certificate".into())
76                })?;
77                let current = migration.assignment();
78                if !outcome.is_commit()
79                    || outcome.epoch != head.epoch
80                    || outcome.checkpoint_id != head.checkpoint_id
81                    || checkpoint.pipeline_identity != migration.descriptor().target_pipeline
82                    || checkpoint.deployment_id != migration.descriptor().deployment_id
83                    || checkpoint.scope != CheckpointScope::Cluster
84                    || assignment.assignment_version
85                        < migration.plan().assignment.assignment_version
86                    || assignment.assignment_version > current.assignment_version
87                    || assignment.assignment_digest != current.assignment_digest
88                    || assignment.vnode_count != current.vnode_count
89                    || assignment.partitioning_abi_version != current.partitioning_abi_version
90                    || !assignment
91                        .participants
92                        .iter()
93                        .map(|participant| participant.node_id)
94                        .eq(current
95                            .participants
96                            .iter()
97                            .map(|participant| participant.node_id))
98                {
99                    return Err(TopologyError::Invalid(
100                        "selected newer checkpoint differs from the committed target/owner map"
101                            .into(),
102                    ));
103                }
104                validate_manifest_budget(&checkpoint)?;
105                Some(Box::new((outcome, checkpoint)))
106            };
107            let fresh = self
108                .committed_topology_restore_input(assignments, processes, operation_id, process)
109                .await?;
110            let after = self.load_record().await?.ok_or(TopologyError::Fenced)?;
111            if !fresh.same_restore_requirements(&migration)
112                || before.commit_head != after.commit_head
113                || before.lease.proof() != after.lease.proof()
114                || after.committed_topology_operation() != Some(operation)
115            {
116                return Err(TopologyError::Fenced);
117            }
118            Ok(TopologyRecoveryInput {
119                migration,
120                target_checkpoint,
121            })
122        })
123        .await
124        .map_err(|_| TopologyError::ReadTimedOut)?
125    }
126}
laminar_core/cluster/control/leader_lease/topology_recovery.rs

laminar_core/cluster/control/leader_lease/
topology_recovery.rs
