laminar_core/cluster/control/leader_lease/
topology_retention.rs

1//! Read exact retained root metadata without reviving ordinary expired checkpoint authority.
2
3use super::topology_admission::{topology_checkpoint_error, CONTROL_TIMEOUT};
4use super::{read_authority_record, CheckpointDecisionStore, LeaderLeaseStore};
5use crate::checkpoint::{CommittedCheckpointIndex, CommittedCheckpointRef};
6use crate::checkpoint_decision::CheckpointOutcome;
7use crate::cluster::control::{TopologyAdmissionStatus, TopologyError};
8
9impl LeaderLeaseStore {
10    /// Return the exact checkpoint roots still pinned by irreversible topology decisions.
11    /// The operation journal bounds this inventory. These references protect state and replay;
12    /// they grant neither assignment authority, actor installation nor input Release.
13    ///
14    /// # Errors
15    /// Rejects missing/corrupt retained evidence, a changed topology/proof or the 15-second budget.
16    pub async fn retained_topology_checkpoints(
17        &self,
18    ) -> Result<Vec<CommittedCheckpointRef>, TopologyError> {
19        tokio::time::timeout(CONTROL_TIMEOUT, async {
20            let before = self.load_record().await?.ok_or(TopologyError::Fenced)?;
21            let mut roots = Vec::new();
22            for operation in before
23                .topology_operations
24                .iter()
25                .filter(|operation| operation.has_target_commit())
26            {
27                Box::pin(self.audit_topology_operation(operation)).await?;
28                let (outcome, _) =
29                    Box::pin(self.load_retained_topology_root_checkpoint(operation)).await?;
30                roots.push(outcome.committed_checkpoint.ok_or(TopologyError::Fenced)?);
31            }
32            let after = self.load_record().await?.ok_or(TopologyError::Fenced)?;
33            if before.lease.proof() != after.lease.proof()
34                || before.topology_operations != after.topology_operations
35            {
36                return Err(TopologyError::Fenced);
37            }
38            roots.sort_by_key(|reference| (reference.epoch, reference.checkpoint_id));
39            roots.dedup();
40            Ok(roots)
41        })
42        .await
43        .map_err(|_| TopologyError::ReadTimedOut)?
44    }
45
46    /// The caller audits the retained operation first. Its immutable cut Commit is the sole
47    /// historical exception to the ordinary artifact floor; GC keeps this exact metadata pinned.
48    pub(super) async fn load_retained_topology_root_checkpoint(
49        &self,
50        operation: &TopologyAdmissionStatus,
51    ) -> Result<(CheckpointOutcome, CommittedCheckpointIndex), TopologyError> {
52        if !operation.has_target_commit() {
53            return Err(TopologyError::Fenced);
54        }
55        let cut = operation.cut.as_ref().ok_or(TopologyError::Fenced)?;
56        let commit = cut.committed.as_ref().ok_or(TopologyError::Fenced)?;
57        let record = read_authority_record(self.store.as_ref(), commit.authority_sequence)
58            .await?
59            .ok_or_else(|| {
60                TopologyError::Invalid("retained root Commit anchor is missing".into())
61            })?;
62        let outcome = record.checkpoint_outcome.ok_or(TopologyError::Fenced)?;
63        if !outcome.is_commit()
64            || outcome.committed_checkpoint.as_ref() != Some(&commit.checkpoint)
65            || outcome.assignment_fence != cut.inventory.assignment_fence
66            || outcome.deployment_id != cut.inventory.deployment_id
67        {
68            return Err(TopologyError::Invalid(
69                "retained root differs from its exact immutable Commit".into(),
70            ));
71        }
72        let index = CheckpointDecisionStore::new(self.store.clone())
73            .validate_committed_checkpoint_for_outcome(&outcome)
74            .await
75            .map_err(|error| topology_checkpoint_error(error.into()))?;
76        if index.pipeline_identity != cut.inventory.pipeline_identity {
77            return Err(TopologyError::Invalid(
78                "retained root differs from its historical pipeline identity".into(),
79            ));
80        }
81        Ok((outcome, index))
82    }
83}
84
85impl LeaderLeaseStore {
86    pub(super) async fn retain_topology_history(
87        &self,
88        head: &super::LeaderAuthorityRecord,
89        retained: &mut std::collections::BTreeSet<u64>,
90    ) -> Result<(), super::LeaseError> {
91        if let Some(baseline) = head.topology_baseline.as_ref() {
92            self.audit_topology_adoption(baseline).await?;
93            retained.insert(baseline.authority_sequence);
94        }
95        for operation in &head.topology_operations {
96            self.audit_topology_operation(operation).await?;
97            retained.insert(operation.admitted_sequence);
98            retained.insert(operation.status_sequence);
99            retained.extend(
100                operation
101                    .target_preparations
102                    .iter()
103                    .map(|receipt| receipt.authority_sequence),
104            );
105            if let Some(commit) = &operation.commit {
106                retained.insert(commit.authority_sequence);
107            }
108            if let Some(activation) = &operation.activation {
109                retained.insert(activation.authority_sequence);
110                retained.extend(
111                    activation
112                        .installations
113                        .iter()
114                        .map(|receipt| receipt.authority_sequence),
115                );
116                if let Some(release) = &activation.release {
117                    retained.insert(release.authority_sequence);
118                }
119            }
120            if let Some(root) = &operation.migration_root {
121                retained.insert(root.authority_sequence);
122            }
123            if let Some(preparation) = &operation.preparation {
124                retained.extend(
125                    preparation
126                        .certificates
127                        .iter()
128                        .map(|certificate| certificate.authority_sequence),
129                );
130            }
131            if let Some(cut) = &operation.cut {
132                retained.insert(cut.bound_sequence);
133                if let Some(commit) = &cut.committed {
134                    retained.insert(commit.authority_sequence);
135                }
136            }
137        }
138        Ok(())
139    }
140}
141
142impl super::LeaderAuthorityRecord {
143    pub(super) fn cleanup_is_pinned(&self, protected: &CommittedCheckpointRef) -> bool {
144        if self.assignment_handoff_pin.as_ref().is_some_and(|pin| {
145            protected.epoch > pin.checkpoint.epoch
146                && protected.checkpoint_id > pin.checkpoint.checkpoint_id
147        }) {
148            return true;
149        }
150        if self.topology_cut_blocks_cleanup(protected) {
151            return true;
152        }
153        if self
154            .outcome_floor
155            .as_ref()
156            .is_some_and(|floor| floor.artifact_before_epoch >= protected.epoch)
157        {
158            return true;
159        }
160        false
161    }
162}
163
164impl LeaderLeaseStore {
165    pub(super) async fn audit_topology_cleanup_roots(&self) -> Result<(), super::DecisionError> {
166        Box::pin(self.retained_topology_checkpoints())
167            .await
168            .map_err(|error| super::DecisionError::Conflict(error.to_string()))?;
169        Ok(())
170    }
171}
laminar_core/cluster/control/leader_lease/topology_retention.rs

laminar_core/cluster/control/leader_lease/
topology_retention.rs
