laminar_core/cluster/control/topology/
checkpoint_continuity.rs

1//! The only explicit parent-to-target checkpoint continuity boundary.
2
3use super::{
4    ClusterTopologyObjectTransition, ClusterTopologyValidation, TopologyError,
5    TopologyMigrationRoot,
6};
7use crate::checkpoint::{CheckpointScope, CommittedCheckpointIndex};
8use crate::cluster::control::CatalogObjectKind;
9
10impl TopologyMigrationRoot {
11    /// Check the first target index against the exact retained parent index and sealed mapping.
12    /// The authority must first audit this root's durable Commit/Release. This pure check grants
13    /// no writer or runtime permission and never changes either checkpoint's historical identity.
14    ///
15    /// # Errors
16    /// Rejects a foreign cut, changed source inventory, ownership/ABI/deployment or regressed
17    /// watermarks. Ordinary checkpoint continuity continues to require identical pipeline IDs.
18    pub fn validate_target_checkpoint_predecessor(
19        &self,
20        descriptor: &ClusterTopologyValidation,
21        target: &CommittedCheckpointIndex,
22        parent: &CommittedCheckpointIndex,
23    ) -> Result<(), TopologyError> {
24        target.validate().map_err(TopologyError::Invalid)?;
25        parent.validate().map_err(TopologyError::Invalid)?;
26        let (_, parent_ref) = parent
27            .encode_and_reference()
28            .map_err(TopologyError::Invalid)?;
29        let parent_assignment = parent
30            .assignment_fence
31            .as_ref()
32            .ok_or(TopologyError::Fenced)?;
33        let target_assignment = target
34            .assignment_fence
35            .as_ref()
36            .ok_or(TopologyError::Fenced)?;
37        let sources = descriptor
38            .objects
39            .iter()
40            .filter(|object| {
41                object.kind == CatalogObjectKind::Source
42                    && object.transition != ClusterTopologyObjectTransition::Remove
43            })
44            .map(|object| object.name.clone())
45            .collect::<Vec<_>>();
46        let parent_sources = descriptor
47            .objects
48            .iter()
49            .filter(|object| {
50                object.kind == CatalogObjectKind::Source
51                    && object.transition != ClusterTopologyObjectTransition::AddFutureOnly
52            })
53            .map(|object| object.name.clone())
54            .collect::<Vec<_>>();
55        if self.cut.checkpoint != parent_ref
56            || target.predecessor.as_ref() != Some(&parent_ref)
57            || target.version < parent.version
58            || target.epoch <= parent.epoch
59            || parent.pipeline_identity != descriptor.parent_pipeline
60            || target.pipeline_identity != descriptor.target_pipeline
61            || parent.deployment_id != descriptor.deployment_id
62            || target.deployment_id != descriptor.deployment_id
63            || parent.scope != CheckpointScope::Cluster
64            || target.scope != CheckpointScope::Cluster
65            || parent.vnode_count != target.vnode_count
66            || parent.source_names != parent_sources
67            || target.source_names != sources
68            || target_assignment.assignment_version < parent_assignment.assignment_version
69            || target_assignment.assignment_digest != parent_assignment.assignment_digest
70            || target_assignment.vnode_count != parent_assignment.vnode_count
71            || target_assignment.partitioning_abi_version
72                != parent_assignment.partitioning_abi_version
73            || !target_assignment
74                .participants
75                .iter()
76                .map(|participant| participant.node_id)
77                .eq(parent_assignment
78                    .participants
79                    .iter()
80                    .map(|participant| participant.node_id))
81        {
82            return Err(TopologyError::Invalid(
83                "first target checkpoint does not continue the exact authorized migration root"
84                    .into(),
85            ));
86        }
87        let inherited_sources = self
88            .preserved_objects
89            .iter()
90            .filter(|object| object.kind == CatalogObjectKind::Source)
91            .map(|object| object.name.clone())
92            .collect::<Vec<_>>();
93        target
94            .validate_source_watermark_continuity(parent, &inherited_sources)
95            .and_then(|()| target.validate_view_frontier_continuity(parent))
96            .map_err(TopologyError::Invalid)
97    }
98}
laminar_core/cluster/control/topology/checkpoint_continuity.rs

laminar_core/cluster/control/topology/
checkpoint_continuity.rs
