laminar_core/cluster/control/leader_lease/
topology_checkpoint.rs

1//! Durable authorization for the first target checkpoint's historical parent predecessor.
2
3use super::topology_admission::CONTROL_TIMEOUT;
4use super::{
5    ClusterCheckpointAuthorityError, DecisionError, LeaderAuthorityRecord, LeaderLeaseStore,
6};
7use crate::checkpoint::CommittedCheckpointIndex;
8use crate::cluster::control::{TopologyAdmissionPhase, TopologyError, TopologyMigrationRoot};
9
10fn transition_error(error: TopologyError) -> ClusterCheckpointAuthorityError {
11    match error {
12        TopologyError::Authority(error) => error.into(),
13        error => DecisionError::Conflict(error.to_string()).into(),
14    }
15}
16
17impl LeaderLeaseStore {
18    /// Validate an exact checkpoint chain edge. A pipeline transition requires the retained,
19    /// audited topology Commit and participant-complete Release for that exact parent root.
20    /// Returns its sealed mapping only for that transition; ordinary edges keep strict equality.
21    /// This is artifact validation, never permission to install actors or open intake.
22    ///
23    /// # Errors
24    /// Rejects unauthorized identity/inventory changes, corrupt/missing migration evidence and
25    /// the 15 second read budget. No fingerprint mismatch override is exposed.
26    pub async fn validate_cluster_checkpoint_predecessor(
27        &self,
28        target: &CommittedCheckpointIndex,
29        parent: &CommittedCheckpointIndex,
30    ) -> Result<Option<TopologyMigrationRoot>, ClusterCheckpointAuthorityError> {
31        if target.pipeline_identity == parent.pipeline_identity {
32            target
33                .validate_predecessor_index(parent)
34                .map_err(DecisionError::Conflict)?;
35            return Ok(None);
36        }
37        tokio::time::timeout(CONTROL_TIMEOUT, async {
38            let current = self
39                .load_record()
40                .await?
41                .ok_or(ClusterCheckpointAuthorityError::Fenced)?;
42            self.validate_checkpoint_predecessor_from(&current, target, parent)
43                .await
44        })
45        .await
46        .map_err(|_| {
47            DecisionError::Conflict("topology checkpoint continuity audit timed out".into())
48        })?
49    }
50
51    pub(super) async fn validate_checkpoint_predecessor_from(
52        &self,
53        current: &LeaderAuthorityRecord,
54        target: &CommittedCheckpointIndex,
55        parent: &CommittedCheckpointIndex,
56    ) -> Result<Option<TopologyMigrationRoot>, ClusterCheckpointAuthorityError> {
57        if target.pipeline_identity == parent.pipeline_identity {
58            target
59                .validate_predecessor_index(parent)
60                .map_err(DecisionError::Conflict)?;
61            return Ok(None);
62        }
63        let (_, parent_ref) = parent
64            .encode_and_reference()
65            .map_err(DecisionError::Conflict)?;
66        let operation = current
67            .topology_operations
68            .iter()
69            .find(|operation| {
70                operation.phase == TopologyAdmissionPhase::Active
71                    && operation.commit.is_some()
72                    && operation
73                        .cut
74                        .as_ref()
75                        .and_then(|cut| cut.committed.as_ref())
76                        .is_some_and(|cut| cut.checkpoint == parent_ref)
77            })
78            .ok_or_else(|| {
79                DecisionError::Conflict(
80                    "checkpoint identity transition has no exact released topology root".into(),
81                )
82            })?;
83        self.audit_topology_operation(operation).await?;
84        let plan = self
85            .load_topology_plan(&operation.plan)
86            .await
87            .map_err(transition_error)?;
88        let descriptor = self
89            .audit_topology_compatibility(&plan)
90            .await
91            .map_err(transition_error)?
92            .ok_or_else(|| {
93                DecisionError::Conflict("topology checkpoint transition has no descriptor".into())
94            })?;
95        let binding = operation.migration_root.as_ref().ok_or_else(|| {
96            DecisionError::Conflict("topology checkpoint transition has no root".into())
97        })?;
98        let root = self
99            .load_topology_root(&binding.root)
100            .await
101            .map_err(transition_error)?;
102        if parent.assignment_fence.as_ref() != Some(&plan.assignment) {
103            return Err(DecisionError::Conflict(
104                "topology checkpoint predecessor differs from its frozen assignment".into(),
105            )
106            .into());
107        }
108        root.validate_target_checkpoint_predecessor(&descriptor, target, parent)
109            .map_err(|error| DecisionError::Conflict(error.to_string()))?;
110        Ok(Some(root))
111    }
112}
laminar_core/cluster/control/leader_lease/topology_checkpoint.rs

laminar_core/cluster/control/leader_lease/
topology_checkpoint.rs
