laminar_core/cluster/control/leader_lease/
topology.rs

1//! Baseline adoption shares the exact authority append with lease/checkpoint/recovery writes.
2
3use std::time::Duration;
4
5use super::{
6    AuthorityCreateOutcome, CheckpointDecisionStore, LeaderAuthorityRecord, LeaderLeaseStore,
7    LeaderProof, LeaseError, AUTHORITY_RECORD_VERSION, TOPOLOGY_ADMISSION_RECORD_VERSION,
8    TOPOLOGY_AUTHORITY_RECORD_VERSION, TOPOLOGY_COMMIT_RECORD_VERSION, TOPOLOGY_CUT_RECORD_VERSION,
9    TOPOLOGY_INSTALLATION_RECORD_VERSION, TOPOLOGY_MIGRATION_ROOT_RECORD_VERSION,
10    TOPOLOGY_PREPARATION_RECORD_VERSION, TOPOLOGY_RECOVERY_RECORD_VERSION,
11    TOPOLOGY_SOURCE_ROOT_RECORD_VERSION, TOPOLOGY_SUBMISSION_RECORD_VERSION,
12    TOPOLOGY_TARGET_PREPARATION_RECORD_VERSION,
13};
14use crate::cluster::control::topology::{
15    LegacyTopologyBaseline, TopologyAdoptionOutcome, TopologyCatalogState, TopologyCommit,
16    TopologyError, TopologyJournalUsage, TopologyOperationId, TopologyVersion,
17    MAX_TOPOLOGY_OPERATIONS, TOPOLOGY_PROTOCOL_VERSION,
18};
19use crate::cluster::control::CatalogManifestRef;
20
21const MAX_TOPOLOGY_CAS_ATTEMPTS: usize = 16;
22const TOPOLOGY_ADOPTION_TIMEOUT: Duration = Duration::from_secs(15);
23const TOPOLOGY_READ_TIMEOUT: Duration = Duration::from_secs(15);
24
25impl LeaderAuthorityRecord {
26    pub(super) fn validate_topology_baseline(&self) -> Result<(), LeaseError> {
27        match (self.version, self.topology_baseline.as_ref()) {
28            (AUTHORITY_RECORD_VERSION, None) => Ok(()),
29            (TOPOLOGY_ADMISSION_RECORD_VERSION, None) if self.topology_operations.is_empty() => {
30                Ok(())
31            }
32            (
33                TOPOLOGY_AUTHORITY_RECORD_VERSION
34                | TOPOLOGY_ADMISSION_RECORD_VERSION
35                | TOPOLOGY_CUT_RECORD_VERSION
36                | TOPOLOGY_PREPARATION_RECORD_VERSION
37                | TOPOLOGY_MIGRATION_ROOT_RECORD_VERSION
38                | TOPOLOGY_SOURCE_ROOT_RECORD_VERSION
39                | TOPOLOGY_TARGET_PREPARATION_RECORD_VERSION
40                | TOPOLOGY_COMMIT_RECORD_VERSION
41                | TOPOLOGY_INSTALLATION_RECORD_VERSION
42                | TOPOLOGY_RECOVERY_RECORD_VERSION
43                | TOPOLOGY_SUBMISSION_RECORD_VERSION,
44                // Installation capability is an explicit coordinated binary format upgrade.
45                Some(baseline),
46            ) => {
47                baseline
48                    .validate()
49                    .map_err(|error| LeaseError::Invalid(error.to_string()))?;
50                if baseline.authority_sequence > self.lease.seq {
51                    return Err(LeaseError::Invalid(
52                        "topology baseline does not bind the sealed catalog and authority sequence"
53                            .into(),
54                    ));
55                }
56                self.validate_committed_topology_chain(baseline)?;
57                Ok(())
58            }
59            _ => Err(LeaseError::Invalid(
60                "authority encoding and topology baseline presence disagree".into(),
61            )),
62        }
63    }
64
65    pub(super) fn validate_topology_successor(&self, next: &Self) -> Result<(), LeaseError> {
66        if let Some(baseline) = self.topology_baseline.as_ref() {
67            if next.version < self.version || next.topology_baseline.as_ref() != Some(baseline) {
68                return Err(LeaseError::Invalid(
69                    "authority append cannot downgrade or replace an adopted topology baseline"
70                        .into(),
71                ));
72            }
73        } else if let Some(baseline) = next.topology_baseline.as_ref() {
74            if !matches!(
75                next.version,
76                TOPOLOGY_AUTHORITY_RECORD_VERSION
77                    | TOPOLOGY_ADMISSION_RECORD_VERSION
78                    | TOPOLOGY_CUT_RECORD_VERSION
79                    | TOPOLOGY_PREPARATION_RECORD_VERSION
80            ) || baseline.authority_sequence != next.lease.seq
81                || self.lease.catalog_manifest.as_ref() != Some(&baseline.manifest)
82            {
83                return Err(LeaseError::Invalid(
84                    "topology adoption must preserve the predecessor inventory at its exact append"
85                        .into(),
86                ));
87            }
88        }
89        self.validate_committed_topology_successor(next)?;
90        Ok(())
91    }
92}
93
94impl LeaderLeaseStore {
95    /// Read explicit topology metadata and validate its referenced inventory/deployment.
96    ///
97    /// Missing version metadata means `LegacySealed`, never an inferred topology version. This
98    /// method never appends a topology operation or initializes a deployment identity. Existing
99    /// authority head reconciliation may finish publication of a previously created record.
100    ///
101    /// # Errors
102    /// Fails closed on malformed, missing or corrupt referenced content or authority.
103    pub async fn topology_catalog_state(&self) -> Result<TopologyCatalogState, TopologyError> {
104        Ok(self
105            .catalog_with_topology()
106            .await?
107            .map_or(TopologyCatalogState::Uninitialized, |(_, state)| state))
108    }
109
110    /// Commit of the newest migration while it still owns startup, assignment and recovery.
111    /// `None` means no committed migration or a settled target: Active with its own committed
112    /// checkpoint, which then follows ordinary startup, reassignment and recovery.
113    ///
114    /// # Errors
115    /// Fails closed on malformed or missing referenced evidence or the bounded read deadline.
116    pub async fn live_topology_commit(&self) -> Result<Option<TopologyCommit>, TopologyError> {
117        tokio::time::timeout(TOPOLOGY_READ_TIMEOUT, async {
118            let Some(head) = self.load_record().await? else {
119                return Ok(None);
120            };
121            let Some(operation) = head.live_topology_migration() else {
122                return Ok(None);
123            };
124            self.audit_topology_operation(operation).await?;
125            Ok(operation.commit.clone())
126        })
127        .await
128        .map_err(|_| TopologyError::ReadTimedOut)?
129    }
130
131    /// Retained topology request identities against the fixed journal capacity.
132    ///
133    /// # Errors
134    /// Fails closed on malformed authority or the bounded read deadline.
135    pub async fn topology_journal_usage(&self) -> Result<TopologyJournalUsage, TopologyError> {
136        tokio::time::timeout(TOPOLOGY_READ_TIMEOUT, async {
137            let retained = self
138                .load_record()
139                .await?
140                .map_or(0, |head| head.topology_operations.len());
141            Ok(TopologyJournalUsage {
142                retained,
143                capacity: MAX_TOPOLOGY_OPERATIONS,
144            })
145        })
146        .await
147        .map_err(|_| TopologyError::ReadTimedOut)?
148    }
149
150    /// Subscription certificates preserved by the newest committed migration. Ordinary startup
151    /// of a settled target binds these exact generations; deriving them from the changed pipeline
152    /// identity would silently start new subscription streams.
153    ///
154    /// # Errors
155    /// Fails closed on malformed or missing referenced evidence or the bounded read deadline.
156    pub async fn committed_topology_subscriptions(
157        &self,
158    ) -> Result<Vec<crate::checkpoint::OutputDistributionCertificate>, TopologyError> {
159        tokio::time::timeout(TOPOLOGY_READ_TIMEOUT, async {
160            let Some(head) = self.load_record().await? else {
161                return Ok(Vec::new());
162            };
163            let Some(operation) = head.committed_topology_operation() else {
164                return Ok(Vec::new());
165            };
166            self.audit_topology_operation(operation).await?;
167            let binding = operation.migration_root.as_ref().ok_or_else(|| {
168                TopologyError::Invalid("committed topology has no migration root".into())
169            })?;
170            let root = self.load_topology_root(&binding.root).await?;
171            Ok(root
172                .subscriptions
173                .into_iter()
174                .map(|subscription| subscription.target_certificate)
175                .collect())
176        })
177        .await
178        .map_err(|_| TopologyError::ReadTimedOut)?
179    }
180
181    pub(in crate::cluster::control) async fn catalog_with_topology(
182        &self,
183    ) -> Result<Option<(super::CatalogManifest, TopologyCatalogState)>, TopologyError> {
184        tokio::time::timeout(TOPOLOGY_READ_TIMEOUT, self.catalog_with_topology_inner())
185            .await
186            .map_err(|_| TopologyError::ReadTimedOut)?
187    }
188
189    async fn catalog_with_topology_inner(
190        &self,
191    ) -> Result<Option<(super::CatalogManifest, TopologyCatalogState)>, TopologyError> {
192        let Some(head) = self.load_record().await? else {
193            return Ok(None);
194        };
195        let Some(manifest) = head.lease.catalog_manifest.as_ref() else {
196            return Ok(None);
197        };
198        let inventory = self.load_catalog_manifest(manifest).await?;
199        let state = match head.topology_baseline.as_ref() {
200            None => TopologyCatalogState::LegacySealed {
201                manifest: manifest.clone(),
202            },
203            Some(baseline) => {
204                self.audit_topology_adoption(baseline).await?;
205                self.require_topology_deployment(&baseline.deployment_id)
206                    .await?;
207                let committed = if let Some(operation) = head.committed_topology_operation() {
208                    self.audit_topology_operation(operation).await?;
209                    operation.commit.clone()
210                } else {
211                    None
212                };
213                TopologyCatalogState::Versioned {
214                    baseline: baseline.clone(),
215                    committed,
216                }
217            }
218        };
219        Ok(Some((inventory, state)))
220    }
221
222    pub(super) async fn audit_topology_adoption(
223        &self,
224        baseline: &LegacyTopologyBaseline,
225    ) -> Result<(), LeaseError> {
226        let record = super::read_authority_record(self.store.as_ref(), baseline.authority_sequence)
227            .await?
228            .ok_or_else(|| LeaseError::Invalid("topology adoption authority is missing".into()))?;
229        if !matches!(
230            record.version,
231            TOPOLOGY_AUTHORITY_RECORD_VERSION
232                | TOPOLOGY_ADMISSION_RECORD_VERSION
233                | TOPOLOGY_CUT_RECORD_VERSION
234                | TOPOLOGY_PREPARATION_RECORD_VERSION
235        ) || record.topology_baseline.as_ref() != Some(baseline)
236        {
237            return Err(LeaseError::Invalid(
238                "topology baseline disagrees with its retained adoption authority".into(),
239            ));
240        }
241        Ok(())
242    }
243
244    pub(super) async fn require_topology_deployment(
245        &self,
246        expected: &str,
247    ) -> Result<(), TopologyError> {
248        let actual = CheckpointDecisionStore::new(self.store.clone())
249            .load_deployment_id()
250            .await
251            .map_err(|error| match error {
252                crate::checkpoint_decision::DecisionError::Io(reason) => {
253                    TopologyError::Authority(LeaseError::Io(reason))
254                }
255                error => TopologyError::Invalid(error.to_string()),
256            })?;
257        if actual.as_deref() != Some(expected) {
258            return Err(TopologyError::Conflict(
259                "topology deployment does not match the existing authority deployment; do not recreate or reset the namespace".into(),
260            ));
261        }
262        Ok(())
263    }
264
265    /// Adopt the exact legacy inventory as topology one, preserving all existing authority.
266    ///
267    /// Call only after a coordinated binary upgrade. The new encoding causes older authority
268    /// readers/writers to fail closed; it does not revoke already-running data-plane actors.
269    /// Adoption itself never changes the graph, opens gates, or authorizes runtime DDL.
270    ///
271    /// A timeout/cancelled request may have committed. Reread `topology_catalog_state` and retry
272    /// the same operation identity/reference. Never infer abort from a lost response.
273    ///
274    /// # Errors
275    /// Rejects a stale leader proof, changed manifest/deployment, unsealed namespace, malformed
276    /// artifacts, storage errors, or bounded deadline/CAS exhaustion.
277    pub async fn adopt_legacy_topology(
278        &self,
279        proof: &LeaderProof,
280        operation_id: TopologyOperationId,
281        expected_manifest: &CatalogManifestRef,
282        expected_deployment: &str,
283    ) -> Result<TopologyAdoptionOutcome, TopologyError> {
284        tokio::time::timeout(
285            TOPOLOGY_ADOPTION_TIMEOUT,
286            self.adopt_legacy_topology_inner(
287                proof,
288                operation_id,
289                expected_manifest,
290                expected_deployment,
291            ),
292        )
293        .await
294        .map_err(|_| TopologyError::Contended)?
295    }
296
297    async fn adopt_legacy_topology_inner(
298        &self,
299        proof: &LeaderProof,
300        operation_id: TopologyOperationId,
301        expected_manifest: &CatalogManifestRef,
302        expected_deployment: &str,
303    ) -> Result<TopologyAdoptionOutcome, TopologyError> {
304        if !proof.is_canonical() {
305            return Err(TopologyError::Fenced);
306        }
307        expected_manifest.validate()?;
308        self.require_topology_deployment(expected_deployment)
309            .await?;
310        let initial = self.load_record().await?.ok_or(TopologyError::Fenced)?;
311        if !initial.lease.matches_proof(proof) {
312            return Err(TopologyError::Fenced);
313        }
314        if initial.lease.catalog_manifest.as_ref() != Some(expected_manifest) {
315            return Err(TopologyError::Conflict(
316                "legacy adoption requires the exact currently sealed manifest".into(),
317            ));
318        }
319        // Validate the original blob before any upgrade append. Do not re-encode or replace it.
320        self.load_catalog_manifest(expected_manifest).await?;
321
322        for _ in 0..MAX_TOPOLOGY_CAS_ATTEMPTS {
323            let published = self
324                .load_published_authority_head()
325                .await?
326                .ok_or(TopologyError::Fenced)?;
327            let current = &published.record;
328            if !current.lease.matches_proof(proof) {
329                return Err(TopologyError::Fenced);
330            }
331            if current.lease.catalog_manifest.as_ref() != Some(expected_manifest) {
332                return Err(TopologyError::Conflict(
333                    "legacy adoption requires the exact currently sealed manifest".into(),
334                ));
335            }
336            if let Some(baseline) = current.topology_baseline.as_ref() {
337                if baseline.deployment_id != expected_deployment {
338                    return Err(TopologyError::Conflict(
339                        "the baseline belongs to another deployment".into(),
340                    ));
341                }
342                self.audit_topology_adoption(baseline).await?;
343                return Ok(TopologyAdoptionOutcome::Existing(baseline.clone()));
344            }
345
346            let sequence = current
347                .lease
348                .seq
349                .checked_add(1)
350                .ok_or_else(|| TopologyError::Invalid("authority sequence exhausted".into()))?;
351            let baseline = LegacyTopologyBaseline {
352                protocol_version: TOPOLOGY_PROTOCOL_VERSION,
353                topology_version: TopologyVersion::LEGACY_BASELINE,
354                manifest: expected_manifest.clone(),
355                deployment_id: expected_deployment.to_owned(),
356                operation_id,
357                authority_sequence: sequence,
358            };
359            baseline.validate()?;
360            let mut lease = current.lease.clone();
361            lease.seq = sequence;
362            let mut candidate = current.preserve_with_lease(lease);
363            candidate.version = candidate.version.max(TOPOLOGY_AUTHORITY_RECORD_VERSION);
364            candidate.topology_baseline = Some(baseline.clone());
365            match self
366                .create_authority_record(Some(&published), &candidate)
367                .await?
368            {
369                AuthorityCreateOutcome::Created => {
370                    return Ok(TopologyAdoptionOutcome::Created(baseline));
371                }
372                AuthorityCreateOutcome::ExistingIdentical => {
373                    return Ok(TopologyAdoptionOutcome::Existing(baseline));
374                }
375                AuthorityCreateOutcome::Contended(winner) => {
376                    if !winner.lease.matches_proof(proof) {
377                        return Err(TopologyError::Fenced);
378                    }
379                    if winner.lease.seq <= current.lease.seq {
380                        return Err(TopologyError::Invalid(
381                            "topology adoption contention did not advance authority".into(),
382                        ));
383                    }
384                    tokio::task::yield_now().await;
385                }
386            }
387        }
388        Err(TopologyError::Contended)
389    }
390}
laminar_core/cluster/control/leader_lease/topology.rs

laminar_core/cluster/control/leader_lease/
topology.rs
