laminar_core/cluster/control/leader_lease/
topology_subscription.rs

1//! Historical subscription certificate continuity through released migration roots.
2
3use super::topology_admission::CONTROL_TIMEOUT;
4use super::{ClusterCheckpointAuthorityError, DecisionError, LeaderLeaseStore};
5use crate::checkpoint::{CommittedCheckpointIndex, OutputDistributionCertificate};
6use crate::cluster::control::{TopologyAdmissionPhase, TopologyError};
7
8fn certificate_error(error: impl std::fmt::Display) -> ClusterCheckpointAuthorityError {
9    DecisionError::Conflict(format!("subscription topology continuity: {error}")).into()
10}
11
12fn topology_error(error: TopologyError) -> ClusterCheckpointAuthorityError {
13    match error {
14        TopologyError::Authority(error) => error.into(),
15        error => certificate_error(error),
16    }
17}
18
19impl LeaderLeaseStore {
20    /// Validate one unchanged subscription incarnation against an exact historical checkpoint.
21    /// A changed pipeline identity requires immutable, participant-complete released roots.
22    /// All other certificate fields retain strict equality. This grants artifact reads only.
23    ///
24    /// # Errors
25    /// Rejects foreign deployment/pipeline identities, missing or corrupt transition evidence,
26    /// changed incarnations/contracts, and the existing 15 second authority read budget.
27    pub async fn validate_cluster_subscription_certificate(
28        &self,
29        index: &CommittedCheckpointIndex,
30        expected: &OutputDistributionCertificate,
31        actual: &OutputDistributionCertificate,
32    ) -> Result<(), ClusterCheckpointAuthorityError> {
33        if actual.pipeline_identity != index.pipeline_identity {
34            return Err(certificate_error(
35                "certificate differs from its checkpoint pipeline",
36            ));
37        }
38        if expected.pipeline_identity == actual.pipeline_identity {
39            return expected.require_match(actual).map_err(certificate_error);
40        }
41        tokio::time::timeout(CONTROL_TIMEOUT, async {
42            let head = self
43                .load_record()
44                .await?
45                .ok_or_else(|| certificate_error("shared authority is missing"))?;
46            if head
47                .topology_baseline
48                .as_ref()
49                .map(|baseline| baseline.deployment_id.as_str())
50                != Some(index.deployment_id.as_str())
51            {
52                return Err(certificate_error(
53                    "checkpoint deployment differs from the adopted baseline",
54                ));
55            }
56            // Both certificates move forward along the same linear committed topology chain.
57            // This supports an old live reader and a new reader replaying older checkpoints
58            // without rewriting either historical manifests or segment authority bindings.
59            let mut left = expected.clone();
60            let mut right = actual.clone();
61            for operation in head.topology_operations.iter().filter(|operation| {
62                operation.phase == TopologyAdmissionPhase::Active && operation.commit.is_some()
63            }) {
64                let plan = self
65                    .load_topology_plan(&operation.plan)
66                    .await
67                    .map_err(topology_error)?;
68                let descriptor = self
69                    .audit_topology_compatibility(&plan)
70                    .await
71                    .map_err(topology_error)?
72                    .ok_or_else(|| {
73                        certificate_error("released topology has no compatibility descriptor")
74                    })?;
75                if left.pipeline_identity != descriptor.parent_pipeline
76                    && right.pipeline_identity != descriptor.parent_pipeline
77                {
78                    continue;
79                }
80                self.audit_topology_operation(operation).await?;
81                let binding = operation
82                    .migration_root
83                    .as_ref()
84                    .ok_or_else(|| certificate_error("released topology has no sealed root"))?;
85                let root = self
86                    .load_topology_root(&binding.root)
87                    .await
88                    .map_err(topology_error)?;
89                if (index.pipeline_identity == descriptor.parent_pipeline
90                    && index.epoch > root.cut.checkpoint.epoch)
91                    || (index.pipeline_identity == descriptor.target_pipeline
92                        && index.epoch <= root.cut.checkpoint.epoch)
93                {
94                    return Err(certificate_error(
95                        "checkpoint lies on the wrong side of its migration cut",
96                    ));
97                }
98                for certificate in [&mut left, &mut right] {
99                    if certificate.pipeline_identity != descriptor.parent_pipeline {
100                        continue;
101                    }
102                    let mapping = root
103                        .subscriptions
104                        .iter()
105                        .find(|mapping| {
106                            mapping.parent_certificate.stream_id == certificate.stream_id
107                        })
108                        .ok_or_else(|| {
109                            certificate_error(
110                                "stream incarnation has no preserved subscription mapping",
111                            )
112                        })?;
113                    certificate
114                        .require_match(&mapping.parent_certificate)
115                        .map_err(certificate_error)?;
116                    certificate.clone_from(&mapping.target_certificate);
117                }
118                if left == right {
119                    return Ok(());
120                }
121            }
122            Err(certificate_error(TopologyError::Conflict(
123                "certificates have no exact retained released topology path".into(),
124            )))
125        })
126        .await
127        .map_err(|_| certificate_error("authority read timed out"))?
128    }
129}
laminar_core/cluster/control/leader_lease/topology_subscription.rs

laminar_core/cluster/control/leader_lease/
topology_subscription.rs
